e2e: more consistent node selection during tests

.github/workflows/release.yml

@@ -2,7 +2,7 @@ name: "Release"
 
 on:
   push:
-    branches:
+    branches: 
       - "RC[0-9]/**"
     tags:
       - "v[0-9]+.[0-9]+.[0-9]+" # Push events to matching v*, i.e. v1.0, v20.15.10
@@ -20,6 +20,9 @@ jobs:
         with:
           go-version: '1.16'
 
+      - run: echo https://github.com/tendermint/tendermint/blob/${GITHUB_REF#refs/tags/}/CHANGELOG.md#${GITHUB_REF#refs/tags/} > ../release_notes.md 
+        if: startsWith(github.ref, 'refs/tags/')
+
       - name: Build
         uses: goreleaser/goreleaser-action@v2
         if: ${{ github.event_name == 'pull_request' }}
@@ -32,6 +35,6 @@ jobs:
         if: startsWith(github.ref, 'refs/tags/')
         with:
           version: latest
-          args: release --rm-dist
+          args: release --rm-dist --release-notes=../release_notes.md
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

CHANGELOG_PENDING.md

@@ -24,10 +24,8 @@ Friendly reminder: We have a [bug bounty program](https://hackerone.com/tendermi
   - [fastsync/rpc] \#6620 Add TotalSyncedTime & RemainingTime to SyncInfo in /status RPC  (@JayT106)
   - [rpc/grpc] \#6725 Mark gRPC in the RPC layer as deprecated.
   - [blockchain/v2] \#6730 Fast Sync v2 is deprecated, please use v0
-  - [rpc] Add genesis_chunked method to support paginated and parallel fetching of large genesis documents.
-  - [rpc/jsonrpc/server] \#6785 `Listen` function updated to take an `int` argument, `maxOpenConnections`, instead of an entire config object. (@williambanfield)
   - [rpc] \#6820 Update RPC methods to reflect changes in the p2p layer, disabling support for `UnsafeDialPeers` and `UnsafeDialPeers` when used with the new p2p layer, and changing the response format of the peer list in `NetInfo` for all users.
-  - [cli] \#6854 Remove deprecated snake case commands. (@tychoish)
+
 - Apps
   - [ABCI] \#6408 Change the `key` and `value` fields from `[]byte` to `string` in the `EventAttribute` type. (@alexanderbez)
   - [ABCI] \#5447 Remove `SetOption` method from `ABCI.Client` interface
@@ -107,7 +105,6 @@ Friendly reminder: We have a [bug bounty program](https://hackerone.com/tendermi
 - [config/indexer] \#6411 Introduce support for custom event indexing data sources, specifically PostgreSQL. (@JayT106)
 - [fastsync/event] \#6619 Emit fastsync status event when switching consensus/fastsync (@JayT106)
 - [statesync/event] \#6700 Emit statesync status start/end event (@JayT106)
-- [inspect] \#6785 Add a new `inspect` command for introspecting the state and block store of a crashed tendermint node. (@williambanfield)
 
 ### IMPROVEMENTS
 
@@ -161,4 +158,4 @@ Friendly reminder: We have a [bug bounty program](https://hackerone.com/tendermi
 - [rpc] \#6507 Ensure RPC client can handle URLs 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)
+- [gitignore] \#6668 Fix gitignore of abci-cli (@tanyabouman)

CONTRIBUTING.md

@@ -328,6 +328,7 @@ If there were no release candidates, begin by creating a backport branch, as des
    - Bump TMVersionDefault version in  `version.go`
    - Bump P2P and block protocol versions in  `version.go`, if necessary
    - Bump ABCI protocol version in `version.go`, if necessary
+   - Add any release notes you would like to be added to the body of the release to `release_notes.md`.
 4. Open a PR with these changes against the backport branch.
 5. Once these changes are on the backport branch, push a tag with prepared release details.
    This will trigger the actual release `v0.35.0`.
@@ -354,6 +355,7 @@ To create a minor release:
    - Bump the ABCI version number, if necessary.
      (Note that ABCI follows semver, and that ABCI versions are the only versions
      which can change during minor releases, and only field additions are valid minor changes.)
+   - Add any release notes you would like to be added to the body of the release to `release_notes.md`.
 4. Open a PR with these changes that will land them back on `v0.35.x`
 5. Once this change has landed on the backport branch, make sure to pull it locally, then push a tag.
    - `git tag -a v0.35.1 -m 'Release v0.35.1'`

cmd/tendermint/commands/gen_node_key.go

@@ -12,9 +12,11 @@ import (
 // GenNodeKeyCmd allows the generation of a node key. It prints JSON-encoded
 // NodeKey to the standard output.
 var GenNodeKeyCmd = &cobra.Command{
-	Use:   "gen-node-key",
-	Short: "Generate a new node key",
-	RunE:  genNodeKey,
+	Use:     "gen-node-key",
+	Aliases: []string{"gen_node_key"},
+	Short:   "Generate a new node key",
+	RunE:    genNodeKey,
+	PreRun:  deprecateSnakeCase,
 }
 
 func genNodeKey(cmd *cobra.Command, args []string) error {

cmd/tendermint/commands/gen_validator.go

@@ -13,9 +13,11 @@ import (
 // GenValidatorCmd allows the generation of a keypair for a
 // validator.
 var GenValidatorCmd = &cobra.Command{
-	Use:   "gen-validator",
-	Short: "Generate new validator keypair",
-	RunE:  genValidator,
+	Use:     "gen-validator",
+	Aliases: []string{"gen_validator"},
+	Short:   "Generate new validator keypair",
+	RunE:    genValidator,
+	PreRun:  deprecateSnakeCase,
 }
 
 func init() {

cmd/tendermint/commands/inspect.go

@@ -1,87 +0,0 @@
-package commands
-
-import (
-	"context"
-	"os"
-	"os/signal"
-	"syscall"
-
-	"github.com/spf13/cobra"
-
-	cfg "github.com/tendermint/tendermint/config"
-	"github.com/tendermint/tendermint/inspect"
-	"github.com/tendermint/tendermint/state"
-	"github.com/tendermint/tendermint/state/indexer/sink"
-	"github.com/tendermint/tendermint/store"
-	"github.com/tendermint/tendermint/types"
-)
-
-// InspectCmd is the command for starting an inspect server.
-var InspectCmd = &cobra.Command{
-	Use:   "inspect",
-	Short: "Run an inspect server for investigating Tendermint state",
-	Long: `
-	inspect runs a subset of Tendermint's RPC endpoints that are useful for debugging
-	issues with Tendermint.
-
-	When the Tendermint consensus engine detects inconsistent state, it will crash the
-	tendermint process. Tendermint will not start up while in this inconsistent state. 
-	The inspect command can be used to query the block and state store using Tendermint
-	RPC calls to debug issues of inconsistent state.
-	`,
-
-	RunE: runInspect,
-}
-
-func init() {
-	InspectCmd.Flags().
-		String("rpc.laddr",
-			config.RPC.ListenAddress, "RPC listenener address. Port required")
-	InspectCmd.Flags().
-		String("db-backend",
-			config.DBBackend, "database backend: goleveldb | cleveldb | boltdb | rocksdb | badgerdb")
-	InspectCmd.Flags().
-		String("db-dir", config.DBPath, "database directory")
-}
-
-func runInspect(cmd *cobra.Command, args []string) error {
-	ctx, cancel := context.WithCancel(cmd.Context())
-	defer cancel()
-
-	c := make(chan os.Signal, 1)
-	signal.Notify(c, syscall.SIGTERM, syscall.SIGINT)
-	go func() {
-		<-c
-		cancel()
-	}()
-
-	blockStoreDB, err := cfg.DefaultDBProvider(&cfg.DBContext{ID: "blockstore", Config: config})
-	if err != nil {
-		return err
-	}
-	blockStore := store.NewBlockStore(blockStoreDB)
-	stateDB, err := cfg.DefaultDBProvider(&cfg.DBContext{ID: "state", Config: config})
-	if err != nil {
-		if err := blockStoreDB.Close(); err != nil {
-			logger.Error("error closing block store db", "error", err)
-		}
-		return err
-	}
-	genDoc, err := types.GenesisDocFromFile(config.GenesisFile())
-	if err != nil {
-		return err
-	}
-	sinks, err := sink.EventSinksFromConfig(config, cfg.DefaultDBProvider, genDoc.ChainID)
-	if err != nil {
-		return err
-	}
-	stateStore := state.NewStore(stateDB)
-
-	ins := inspect.New(config.RPC, blockStore, stateStore, sinks, logger)
-
-	logger.Info("starting inspect server")
-	if err := ins.Run(ctx); err != nil {
-		return err
-	}
-	return nil
-}

cmd/tendermint/commands/probe_upnp.go

@@ -11,9 +11,11 @@ import (
 
 // ProbeUpnpCmd adds capabilities to test the UPnP functionality.
 var ProbeUpnpCmd = &cobra.Command{
-	Use:   "probe-upnp",
-	Short: "Test UPnP functionality",
-	RunE:  probeUpnp,
+	Use:     "probe-upnp",
+	Aliases: []string{"probe_upnp"},
+	Short:   "Test UPnP functionality",
+	RunE:    probeUpnp,
+	PreRun:  deprecateSnakeCase,
 }
 
 func probeUpnp(cmd *cobra.Command, args []string) error {

cmd/tendermint/commands/replay.go

@@ -17,9 +17,11 @@ var ReplayCmd = &cobra.Command{
 // ReplayConsoleCmd allows replaying of messages from the WAL in a
 // console.
 var ReplayConsoleCmd = &cobra.Command{
-	Use:   "replay-console",
-	Short: "Replay messages from WAL in a console",
+	Use:     "replay-console",
+	Aliases: []string{"replay_console"},
+	Short:   "Replay messages from WAL in a console",
 	Run: func(cmd *cobra.Command, args []string) {
 		consensus.RunReplayFile(config.BaseConfig, config.Consensus, true)
 	},
+	PreRun: deprecateSnakeCase,
 }

cmd/tendermint/commands/reset_priv_validator.go

@@ -14,9 +14,11 @@ import (
 // ResetAllCmd removes the database of this Tendermint core
 // instance.
 var ResetAllCmd = &cobra.Command{
-	Use:   "unsafe-reset-all",
-	Short: "(unsafe) Remove all the data and WAL, reset this node's validator to genesis state",
-	RunE:  resetAll,
+	Use:     "unsafe-reset-all",
+	Aliases: []string{"unsafe_reset_all"},
+	Short:   "(unsafe) Remove all the data and WAL, reset this node's validator to genesis state",
+	RunE:    resetAll,
+	PreRun:  deprecateSnakeCase,
 }
 
 var keepAddrBook bool
@@ -29,9 +31,11 @@ func init() {
 
 // ResetPrivValidatorCmd resets the private validator files.
 var ResetPrivValidatorCmd = &cobra.Command{
-	Use:   "unsafe-reset-priv-validator",
-	Short: "(unsafe) Reset this node's validator to genesis state",
-	RunE:  resetPrivValidator,
+	Use:     "unsafe-reset-priv-validator",
+	Aliases: []string{"unsafe_reset_priv_validator"},
+	Short:   "(unsafe) Reset this node's validator to genesis state",
+	RunE:    resetPrivValidator,
+	PreRun:  deprecateSnakeCase,
 }
 
 // XXX: this is totally unsafe.

cmd/tendermint/commands/root.go

@@ -2,6 +2,7 @@ package commands
 
 import (
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/spf13/cobra"
@@ -64,3 +65,10 @@ var RootCmd = &cobra.Command{
 		return nil
 	},
 }
+
+// deprecateSnakeCase is a util function for 0.34.1. Should be removed in 0.35
+func deprecateSnakeCase(cmd *cobra.Command, args []string) {
+	if strings.Contains(cmd.CalledAs(), "_") {
+		fmt.Println("Deprecated: snake_case commands will be replaced by hyphen-case commands in the next major release")
+	}
+}

cmd/tendermint/commands/show_node_id.go

@@ -8,9 +8,11 @@ import (
 
 // ShowNodeIDCmd dumps node's ID to the standard output.
 var ShowNodeIDCmd = &cobra.Command{
-	Use:   "show-node-id",
-	Short: "Show this node's ID",
-	RunE:  showNodeID,
+	Use:     "show-node-id",
+	Aliases: []string{"show_node_id"},
+	Short:   "Show this node's ID",
+	RunE:    showNodeID,
+	PreRun:  deprecateSnakeCase,
 }
 
 func showNodeID(cmd *cobra.Command, args []string) error {

cmd/tendermint/commands/show_validator.go

@@ -16,9 +16,11 @@ import (
 
 // ShowValidatorCmd adds capabilities for showing the validator info.
 var ShowValidatorCmd = &cobra.Command{
-	Use:   "show-validator",
-	Short: "Show this node's validator info",
-	RunE:  showValidator,
+	Use:     "show-validator",
+	Aliases: []string{"show_validator"},
+	Short:   "Show this node's validator info",
+	RunE:    showValidator,
+	PreRun:  deprecateSnakeCase,
 }
 
 func showValidator(cmd *cobra.Command, args []string) error {

cmd/tendermint/main.go

@@ -28,7 +28,6 @@ func main() {
 		cmd.ShowNodeIDCmd,
 		cmd.GenNodeKeyCmd,
 		cmd.VersionCmd,
-		cmd.InspectCmd,
 		cmd.MakeKeyMigrateCommand(),
 		debug.DebugCmd,
 		cli.NewCompletionCmd(rootCmd, true),

docs/architecture/README.md

@@ -98,5 +98,3 @@ Note the context/background should be written in the present tense.
 - [ADR-045: ABCI-Evidence](./adr-045-abci-evidence.md)
 - [ADR-057: RPC](./adr-057-RPC.md)
 - [ADR-069: Node Initialization](./adr-069-flexible-node-initialization.md)
-- [ADR-071: Proposer-Based Timestamps](adr-071-proposer-based-timestamps.md)
-- [ADR-072: Restore Requests for Comments](./adr-072-request-for-comments.md)

docs/architecture/adr-072-request-for-comments.md

@@ -1,105 +0,0 @@
-# ADR 72: Restore Requests for Comments
-
-## Changelog
-
-- 20-Aug-2021: Initial draft (@creachadair)
-
-## Status
-
-Proposed
-
-## Context
-
-In the past, we kept a collection of Request for Comments (RFC) documents in `docs/rfc`.
-Prior to the creation of the ADR process, these documents were used to document
-design and implementation decisions about Tendermint Core. The RFC directory
-was removed in favor of ADRs, in commit 3761aa69 (PR
-[\#6345](https://github.com/tendermint/tendermint/pull/6345)).
-
-For issues where an explicit design decision or implementation change is
-required, an ADR is generally preferable to an open-ended RFC: An ADR is
-relatively narrowly-focused, identifies a specific design or implementation
-question, and documents the consensus answer to that question.
-
-Some discussions are more open-ended, however, or don't require a specific
-decision to be made (yet). Such conversations are still valuable to document,
-and several members of the Tendermint team have been doing so by writing gists
-or Google docs to share them around. That works well enough in the moment, but
-gists do not support any kind of collaborative editing, and both gists and docs
-are hard to discover after the fact. Google docs have much better collaborative
-editing, but are worse for discoverability, especially when contributors span
-different Google accounts.
-
-Discoverability is important, because these kinds of open-ended discussions are
-useful to people who come later -- either as new team members or as outside
-contributors seeking to use and understand the thoughts behind our designs and
-the architectural decisions that arose from those discussion.
-
-With these in mind, I propose that:
-
--  We re-create a new, initially empty `docs/rfc` directory in the repository,
-   and use it to capture these kinds of open-ended discussions in supplement to
-   ADRs.
-
--  Unlike in the previous RFC scheme, documents in this new directory will
-   _not_ be used directly for decision-making. This is the key difference
-   between an RFC and an ADR.
-
-   Instead, an RFC will exist to document background, articulate general
-   principles, and serve as a historical record of discussion and motivation.
-
-   In this system, an RFC may _only_ result in a decision indirectly, via ADR
-   documents created in response to the RFC.
-
-   **In short:** If a decision is required, write an ADR; otherwise if a
-   sufficiently broad discussion is needed, write an RFC.
-
-Just so that there is a consistent format, I also propose that:
-
--  RFC files are named `rfc-XXX-title.{md,rst,txt}` and are written in plain
-   text, Markdown, or ReStructured Text.
-
--  Like an ADR, an RFC should include a high-level change log at the top of the
-   document, and sections for:
-
-     * Abstract: A brief, high-level synopsis of the topic.
-     * Background: Any background necessary to understand the topic.
-     * Discussion: Detailed discussion of the issue being considered.
-
--  Unlike an ADR, an RFC does _not_ include sections for Decisions, Detailed
-   Design, or evaluation of proposed solutions. If an RFC leads to a proposal
-   for an actual architectural change, that must be recorded in an ADR in the
-   usual way, and may refer back to the RFC in its References section.
-
-## Alternative Approaches
-
-Leaving aside implementation details, the main alternative to this proposal is
-to leave things as they are now, with ADRs as the only log of record and other
-discussions being held informally in whatever medium is convenient at the time.
-
-## Decision
-
-(pending)
-
-## Detailed Design
-
-- Create a new `docs/rfc` directory in the `tendermint` repository. Note that
-  this proposal intentionally does _not_ pull back the previous contents of
-  that path from Git history, as those documents were appropriately merged into
-  the ADR process.
-
-- Create a `README.md` for RFCs that explains the rules and their relationship
-  to ADRs.
-
-- Create an `rfc-template.md` file for RFC files.
-
-## Consequences
-
-### Positive
-
-- We will have a more discoverable place to record open-ended discussions that
-  do not immediately result in a design change.
-
-### Negative
-
-- Potentially some people could be confused about the RFC/ADR distinction.

docs/rfc/README.md

@@ -1,40 +0,0 @@
----
-order: 1
-parent:
-  order: false
----
-
-# Requests for Comments
-
-A Request for Comments (RFC) is a record of discussion on an open-ended topic
-related to the design and implementation of Tendermint Core, for which no
-immediate decision is required.
-
-The purpose of an RFC is to serve as a historical record of a high-level
-discussion that might otherwise only be recorded in an ad hoc way (for example,
-via gists or Google docs) that are difficult to discover for someone after the
-fact. An RFC _may_ give rise to more specific architectural _decisions_ for
-Tendermint, but those decisions must be recorded separately in [Architecture
-Decision Records (ADR)](./../architecture).
-
-As a rule of thumb, if you can articulate a specific question that needs to be
-answered, write an ADR. If you need to explore the topic and get input from
-others to know what questions need to be answered, an RFC may be appropriate.
-
-## RFC Content
-
-An RFC should provide:
-
-- A **changelog**, documenting when and how the RFC has changed.
-- An **abstract**, briefly summarizing the topic so the reader can quickly tell
-  whether it is relevant to their interest.
-- Any **background** a reader will need to understand and participate in the
-  substance of the discussion (links to other documents are fine here).
-- The **discussion**, the primary content of the document.
-
-The [rfc-template.md](./rfc-template.md) file includes placeholders for these
-sections.
-
-## Table of Contents
-
-<!-- - [RFC-NNN: Title](./rfc-NNN.title.md) -->

docs/rfc/rfc-template.md

@@ -1,35 +0,0 @@
-# RFC {RFC-NUMBER}: {TITLE}
-
-## Changelog
-
-- {date}: {changelog}
-
-## Abstract
-
-> A brief high-level synopsis of the topic of discussion for this RFC, ideally
-> just a few sentences.  This should help the reader quickly decide whether the
-> rest of the discussion is relevant to their interest.
-
-## Background
-
-> Any context or orientation needed for a reader to understand and participate
-> in the substance of the Discussion. If necessary, this section may include
-> links to other documentation or sources rather than restating existing
-> material, but should provide enough detail that the reader can tell what they
-> need to read to be up-to-date.
-
-### References
-
-> Links to external materials needed to follow the discussion may be added here.
->
-> In addition, if the discussion in a request for comments leads to any design
-> decisions, it may be helpful to add links to the ADR documents here after the
-> discussion has settled.
-
-## Discussion
-
-> This section contains the core of the discussion.
->
-> There is no fixed format for this section, but ideally changes to this
-> section should be updated before merging to reflect any discussion that took
-> place on the PR that made those changes.

docs/tools/debugging/README.md

@@ -62,30 +62,3 @@ given destination directory. Each archive will contain:
 
 Note: goroutine.out and heap.out will only be written if a profile address is
 provided and is operational. This command is blocking and will log any error.
-
-## Tendermint Inspect
-
-Tendermint includes an `inspect` command for querying Tendermint's state store and block
-store over Tendermint RPC.
-
-When the Tendermint consensus engine detects inconsistent state, it will crash the
-entire Tendermint process. 
-While in this inconsistent state, a node running Tendermint's consensus engine will not start up. 
-The `inspect` command runs only a subset of Tendermint's RPC endpoints for querying the block store
-and state store. 
-`inspect` allows operators to query a read-only view of the stage.
-`inspect` does not run the consensus engine at all and can therefore be used to debug
-processes that have crashed due to inconsistent state. 
-
-
-To start the `inspect` process, run
-```bash
-tendermint inspect
-```
-
-### RPC endpoints
-The list of available RPC endpoints can be found by making a request to the RPC port.
-For an `inspect` process running on `127.0.0.1:26657`, navigate your browser to 
-`http://127.0.0.1:26657/` to retrieve the list of enabled RPC endpoints.
-
-Additional information on the Tendermint RPC endpoints can be found in the [rpc documentation](https://docs.tendermint.com/master/rpc).

docs/tools/debugging/pro.md

@@ -64,42 +64,13 @@ It won’t kill the node, but it will gather all of the above data and package i
 
 At this point, depending on how severe the degradation is, you may want to restart the process.
 
-## Tendermint Inspect
-
-What if the Tendermint node will not start up due to inconsistent consensus state? 
-
-When a node running the Tendermint consensus engine detects an inconsistent state 
-it will crash the entire Tendermint process. 
-The Tendermint consensus engine cannot be run in this inconsistent state and the so node
-will fail to start up as a result.
-The Tendermint RPC server can provide valuable information for debugging in this situation.
-The Tendermint `inspect` command will run a subset of the Tendermint RPC server 
-that is useful for debugging inconsistent state.
-
-### Running inspect
-
-Start up the `inspect` tool on the machine where Tendermint crashed using: 
-```bash
-tendermint inspect --home=</path/to/app.d>
-```
-
-`inspect` will use the data directory specified in your Tendermint configuration file.
-`inspect` will also run the RPC server at the address specified in your Tendermint configuration file.
-
-### Using inspect
-
-With the `inspect` server running, you can access RPC endpoints that are critically important
-for debugging.
-Calling the `/status`, `/consensus_state` and `/dump_consensus_state` RPC endpoint 
-will return useful information about the Tendermint consensus state.
-
 ## Outro
 
-We’re hoping that these Tendermint tools will become de facto the first response for any accidents.
+We’re hoping that the `tendermint debug` subcommand will become de facto the first response to any accidents.
 
-Let us know what your experience has been so far! Have you had a chance to try `tendermint debug` or `tendermint inspect` yet?
+Let us know what your experience has been so far! Have you had a chance to try `tendermint debug` yet?
 
-Join our [discord chat](https://discord.gg/vcExX9T), where we discuss the current issues and future improvements.
+Join our chat, where we discuss the current issues and future improvements.
 
 —
 

go.mod

@@ -37,7 +37,6 @@ require (
 	github.com/vektra/mockery/v2 v2.9.0
 	golang.org/x/crypto v0.0.0-20210513164829-c07d793c2f9a
 	golang.org/x/net v0.0.0-20210428140749-89ef3d95e781
-	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	google.golang.org/grpc v1.40.0
 	gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b // indirect
 	pgregory.net/rapid v0.4.7

go.sum

@@ -1075,7 +1075,6 @@ golang.org/x/sync v0.0.0-20200317015054-43a5402ce75a/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20200625203802-6e8e738ad208/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
-golang.org/x/sync v0.0.0-20210220032951-036812b2e83c h1:5KslGYwFpkhGh+Q16bwMP3cOontH8FOep7tGV86Y7SQ=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

inspect/doc.go

@@ -1,36 +0,0 @@
-/*
-Package inspect provides a tool for investigating the state of a
-failed Tendermint node.
-
-This package provides the Inspector type. The Inspector type runs a subset of the Tendermint
-RPC endpoints that are useful for debugging issues with Tendermint consensus.
-
-When a node running the Tendermint consensus engine detects an inconsistent consensus state,
-the entire node will crash. The Tendermint consensus engine cannot run in this
-inconsistent state so the node will not be able to start up again.
-
-The RPC endpoints provided by the Inspector type allow for a node operator to inspect
-the block store and state store to better understand what may have caused the inconsistent state.
-
-
-The Inspector type's lifecycle is controlled by a context.Context
-  ins := inspect.NewFromConfig(rpcConfig)
-  ctx, cancelFunc:= context.WithCancel(context.Background())
-
-  // Run blocks until the Inspector server is shut down.
-  go ins.Run(ctx)
-  ...
-
-  // calling the cancel function will stop the running inspect server
-  cancelFunc()
-
-Inspector serves its RPC endpoints on the address configured in the RPC configuration
-
-  rpcConfig.ListenAddress = "tcp://127.0.0.1:26657"
-  ins := inspect.NewFromConfig(rpcConfig)
-  go ins.Run(ctx)
-
-The list of available RPC endpoints can then be viewed by navigating to
-http://127.0.0.1:26657/ in the web browser.
-*/
-package inspect

inspect/inspect.go

@@ -1,149 +0,0 @@
-package inspect
-
-import (
-	"context"
-	"errors"
-	"fmt"
-	"net"
-
-	"github.com/tendermint/tendermint/config"
-	"github.com/tendermint/tendermint/inspect/rpc"
-	"github.com/tendermint/tendermint/libs/log"
-	tmstrings "github.com/tendermint/tendermint/libs/strings"
-	rpccore "github.com/tendermint/tendermint/rpc/core"
-	"github.com/tendermint/tendermint/state"
-	"github.com/tendermint/tendermint/state/indexer"
-	"github.com/tendermint/tendermint/state/indexer/sink"
-	"github.com/tendermint/tendermint/store"
-	"github.com/tendermint/tendermint/types"
-
-	"golang.org/x/sync/errgroup"
-)
-
-// Inspector manages an RPC service that exports methods to debug a failed node.
-// After a node shuts down due to a consensus failure, it will no longer start
-// up its state cannot easily be inspected. An Inspector value provides a similar interface
-// to the node, using the underlying Tendermint data stores, without bringing up
-// any other components. A caller can query the Inspector service to inspect the
-// persisted state and debug the failure.
-type Inspector struct {
-	routes rpccore.RoutesMap
-
-	config *config.RPCConfig
-
-	indexerService *indexer.Service
-	eventBus       *types.EventBus
-	logger         log.Logger
-}
-
-// New returns an Inspector that serves RPC on the specified BlockStore and StateStore.
-// The Inspector type does not modify the state or block stores.
-// The sinks are used to enable block and transaction querying via the RPC server.
-// The caller is responsible for starting and stopping the Inspector service.
-///
-//nolint:lll
-func New(cfg *config.RPCConfig, bs state.BlockStore, ss state.Store, es []indexer.EventSink, logger log.Logger) *Inspector {
-	routes := rpc.Routes(*cfg, ss, bs, es, logger)
-	eb := types.NewEventBus()
-	eb.SetLogger(logger.With("module", "events"))
-	is := indexer.NewIndexerService(es, eb)
-	is.SetLogger(logger.With("module", "txindex"))
-	return &Inspector{
-		routes:         routes,
-		config:         cfg,
-		logger:         logger,
-		eventBus:       eb,
-		indexerService: is,
-	}
-}
-
-// NewFromConfig constructs an Inspector using the values defined in the passed in config.
-func NewFromConfig(cfg *config.Config) (*Inspector, error) {
-	bsDB, err := config.DefaultDBProvider(&config.DBContext{ID: "blockstore", Config: cfg})
-	if err != nil {
-		return nil, err
-	}
-	bs := store.NewBlockStore(bsDB)
-	sDB, err := config.DefaultDBProvider(&config.DBContext{ID: "state", Config: cfg})
-	if err != nil {
-		return nil, err
-	}
-	genDoc, err := types.GenesisDocFromFile(cfg.GenesisFile())
-	if err != nil {
-		return nil, err
-	}
-	sinks, err := sink.EventSinksFromConfig(cfg, config.DefaultDBProvider, genDoc.ChainID)
-	if err != nil {
-		return nil, err
-	}
-	logger := log.MustNewDefaultLogger(log.LogFormatPlain, log.LogLevelInfo, false)
-	ss := state.NewStore(sDB)
-	return New(cfg.RPC, bs, ss, sinks, logger), nil
-}
-
-// Run starts the Inspector servers and blocks until the servers shut down. The passed
-// in context is used to control the lifecycle of the servers.
-func (ins *Inspector) Run(ctx context.Context) error {
-	err := ins.eventBus.Start()
-	if err != nil {
-		return fmt.Errorf("error starting event bus: %s", err)
-	}
-	defer func() {
-		err := ins.eventBus.Stop()
-		if err != nil {
-			ins.logger.Error("event bus stopped with error", "err", err)
-		}
-	}()
-	err = ins.indexerService.Start()
-	if err != nil {
-		return fmt.Errorf("error starting indexer service: %s", err)
-	}
-	defer func() {
-		err := ins.indexerService.Stop()
-		if err != nil {
-			ins.logger.Error("indexer service stopped with error", "err", err)
-		}
-	}()
-	return startRPCServers(ctx, ins.config, ins.logger, ins.routes)
-}
-
-func startRPCServers(ctx context.Context, cfg *config.RPCConfig, logger log.Logger, routes rpccore.RoutesMap) error {
-	g, tctx := errgroup.WithContext(ctx)
-	listenAddrs := tmstrings.SplitAndTrimEmpty(cfg.ListenAddress, ",", " ")
-	rh := rpc.Handler(cfg, routes, logger)
-	for _, listenerAddr := range listenAddrs {
-		server := rpc.Server{
-			Logger:  logger,
-			Config:  cfg,
-			Handler: rh,
-			Addr:    listenerAddr,
-		}
-		if cfg.IsTLSEnabled() {
-			keyFile := cfg.KeyFile()
-			certFile := cfg.CertFile()
-			listenerAddr := listenerAddr
-			g.Go(func() error {
-				logger.Info("RPC HTTPS server starting", "address", listenerAddr,
-					"certfile", certFile, "keyfile", keyFile)
-				err := server.ListenAndServeTLS(tctx, certFile, keyFile)
-				if !errors.Is(err, net.ErrClosed) {
-					return err
-				}
-				logger.Info("RPC HTTPS server stopped", "address", listenerAddr)
-				return nil
-			})
-		} else {
-			listenerAddr := listenerAddr
-			g.Go(func() error {
-				logger.Info("RPC HTTP server starting", "address", listenerAddr)
-				err := server.ListenAndServe(tctx)
-				if !errors.Is(err, net.ErrClosed) {
-					return err
-				}
-				logger.Info("RPC HTTP server stopped", "address", listenerAddr)
-				return nil
-			})
-		}
-	}
-	return g.Wait()
-}

inspect/inspect_test.go

@@ -1,583 +0,0 @@
-package inspect_test
-
-import (
-	"context"
-	"fmt"
-	"net"
-	"os"
-	"strings"
-	"sync"
-	"testing"
-	"time"
-
-	"github.com/fortytw2/leaktest"
-	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
-	abcitypes "github.com/tendermint/tendermint/abci/types"
-	"github.com/tendermint/tendermint/config"
-	"github.com/tendermint/tendermint/inspect"
-	"github.com/tendermint/tendermint/libs/log"
-	"github.com/tendermint/tendermint/libs/pubsub/query"
-	"github.com/tendermint/tendermint/proto/tendermint/state"
-	httpclient "github.com/tendermint/tendermint/rpc/client/http"
-	"github.com/tendermint/tendermint/state/indexer"
-	indexermocks "github.com/tendermint/tendermint/state/indexer/mocks"
-	statemocks "github.com/tendermint/tendermint/state/mocks"
-	"github.com/tendermint/tendermint/types"
-)
-
-func TestInspectConstructor(t *testing.T) {
-	cfg := config.ResetTestRoot("test")
-	t.Cleanup(leaktest.Check(t))
-	defer func() { _ = os.RemoveAll(cfg.RootDir) }()
-	t.Run("from config", func(t *testing.T) {
-		d, err := inspect.NewFromConfig(cfg)
-		require.NoError(t, err)
-		require.NotNil(t, d)
-	})
-
-}
-
-func TestInspectRun(t *testing.T) {
-	cfg := config.ResetTestRoot("test")
-	t.Cleanup(leaktest.Check(t))
-	defer func() { _ = os.RemoveAll(cfg.RootDir) }()
-	t.Run("from config", func(t *testing.T) {
-		d, err := inspect.NewFromConfig(cfg)
-		require.NoError(t, err)
-		ctx, cancel := context.WithCancel(context.Background())
-		stoppedWG := &sync.WaitGroup{}
-		stoppedWG.Add(1)
-		go func() {
-			require.NoError(t, d.Run(ctx))
-			stoppedWG.Done()
-		}()
-		cancel()
-		stoppedWG.Wait()
-	})
-
-}
-
-func TestBlock(t *testing.T) {
-	testHeight := int64(1)
-	testBlock := new(types.Block)
-	testBlock.Header.Height = testHeight
-	testBlock.Header.LastCommitHash = []byte("test hash")
-	stateStoreMock := &statemocks.Store{}
-
-	blockStoreMock := &statemocks.BlockStore{}
-	blockStoreMock.On("Height").Return(testHeight)
-	blockStoreMock.On("Base").Return(int64(0))
-	blockStoreMock.On("LoadBlockMeta", testHeight).Return(&types.BlockMeta{})
-	blockStoreMock.On("LoadBlock", testHeight).Return(testBlock)
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-	resultBlock, err := cli.Block(context.Background(), &testHeight)
-	require.NoError(t, err)
-	require.Equal(t, testBlock.Height, resultBlock.Block.Height)
-	require.Equal(t, testBlock.LastCommitHash, resultBlock.Block.LastCommitHash)
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func TestTxSearch(t *testing.T) {
-	testHash := []byte("test")
-	testTx := []byte("tx")
-	testQuery := fmt.Sprintf("tx.hash='%s'", string(testHash))
-	testTxResult := &abcitypes.TxResult{
-		Height: 1,
-		Index:  100,
-		Tx:     testTx,
-	}
-
-	stateStoreMock := &statemocks.Store{}
-	blockStoreMock := &statemocks.BlockStore{}
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	eventSinkMock.On("Type").Return(indexer.KV)
-	eventSinkMock.On("SearchTxEvents", mock.Anything,
-		mock.MatchedBy(func(q *query.Query) bool { return testQuery == q.String() })).
-		Return([]*abcitypes.TxResult{testTxResult}, nil)
-
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-
-	var page = 1
-	resultTxSearch, err := cli.TxSearch(context.Background(), testQuery, false, &page, &page, "")
-	require.NoError(t, err)
-	require.Len(t, resultTxSearch.Txs, 1)
-	require.Equal(t, types.Tx(testTx), resultTxSearch.Txs[0].Tx)
-
-	cancel()
-	wg.Wait()
-
-	eventSinkMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-	blockStoreMock.AssertExpectations(t)
-}
-func TestTx(t *testing.T) {
-	testHash := []byte("test")
-	testTx := []byte("tx")
-
-	stateStoreMock := &statemocks.Store{}
-	blockStoreMock := &statemocks.BlockStore{}
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	eventSinkMock.On("Type").Return(indexer.KV)
-	eventSinkMock.On("GetTxByHash", testHash).Return(&abcitypes.TxResult{
-		Tx: testTx,
-	}, nil)
-
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-
-	res, err := cli.Tx(context.Background(), testHash, false)
-	require.NoError(t, err)
-	require.Equal(t, types.Tx(testTx), res.Tx)
-
-	cancel()
-	wg.Wait()
-
-	eventSinkMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-	blockStoreMock.AssertExpectations(t)
-}
-func TestConsensusParams(t *testing.T) {
-	testHeight := int64(1)
-	testMaxGas := int64(55)
-	stateStoreMock := &statemocks.Store{}
-	blockStoreMock := &statemocks.BlockStore{}
-	blockStoreMock.On("Height").Return(testHeight)
-	blockStoreMock.On("Base").Return(int64(0))
-	stateStoreMock.On("LoadConsensusParams", testHeight).Return(types.ConsensusParams{
-		Block: types.BlockParams{
-			MaxGas: testMaxGas,
-		},
-	}, nil)
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-	params, err := cli.ConsensusParams(context.Background(), &testHeight)
-	require.NoError(t, err)
-	require.Equal(t, params.ConsensusParams.Block.MaxGas, testMaxGas)
-
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func TestBlockResults(t *testing.T) {
-	testHeight := int64(1)
-	testGasUsed := int64(100)
-	stateStoreMock := &statemocks.Store{}
-	//	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
-	stateStoreMock.On("LoadABCIResponses", testHeight).Return(&state.ABCIResponses{
-		DeliverTxs: []*abcitypes.ResponseDeliverTx{
-			{
-				GasUsed: testGasUsed,
-			},
-		},
-		EndBlock:   &abcitypes.ResponseEndBlock{},
-		BeginBlock: &abcitypes.ResponseBeginBlock{},
-	}, nil)
-	blockStoreMock := &statemocks.BlockStore{}
-	blockStoreMock.On("Base").Return(int64(0))
-	blockStoreMock.On("Height").Return(testHeight)
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-	res, err := cli.BlockResults(context.Background(), &testHeight)
-	require.NoError(t, err)
-	require.Equal(t, res.TotalGasUsed, testGasUsed)
-
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func TestCommit(t *testing.T) {
-	testHeight := int64(1)
-	testRound := int32(101)
-	stateStoreMock := &statemocks.Store{}
-	blockStoreMock := &statemocks.BlockStore{}
-	blockStoreMock.On("Base").Return(int64(0))
-	blockStoreMock.On("Height").Return(testHeight)
-	blockStoreMock.On("LoadBlockMeta", testHeight).Return(&types.BlockMeta{}, nil)
-	blockStoreMock.On("LoadSeenCommit").Return(&types.Commit{
-		Height: testHeight,
-		Round:  testRound,
-	}, nil)
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-	res, err := cli.Commit(context.Background(), &testHeight)
-	require.NoError(t, err)
-	require.NotNil(t, res)
-	require.Equal(t, res.SignedHeader.Commit.Round, testRound)
-
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func TestBlockByHash(t *testing.T) {
-	testHeight := int64(1)
-	testHash := []byte("test hash")
-	testBlock := new(types.Block)
-	testBlock.Header.Height = testHeight
-	testBlock.Header.LastCommitHash = testHash
-	stateStoreMock := &statemocks.Store{}
-	blockStoreMock := &statemocks.BlockStore{}
-	blockStoreMock.On("LoadBlockMeta", testHeight).Return(&types.BlockMeta{
-		BlockID: types.BlockID{
-			Hash: testHash,
-		},
-		Header: types.Header{
-			Height: testHeight,
-		},
-	}, nil)
-	blockStoreMock.On("LoadBlockByHash", testHash).Return(testBlock, nil)
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-	res, err := cli.BlockByHash(context.Background(), testHash)
-	require.NoError(t, err)
-	require.NotNil(t, res)
-	require.Equal(t, []byte(res.BlockID.Hash), testHash)
-
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func TestBlockchain(t *testing.T) {
-	testHeight := int64(1)
-	testBlock := new(types.Block)
-	testBlockHash := []byte("test hash")
-	testBlock.Header.Height = testHeight
-	testBlock.Header.LastCommitHash = testBlockHash
-	stateStoreMock := &statemocks.Store{}
-
-	blockStoreMock := &statemocks.BlockStore{}
-	blockStoreMock.On("Height").Return(testHeight)
-	blockStoreMock.On("Base").Return(int64(0))
-	blockStoreMock.On("LoadBlockMeta", testHeight).Return(&types.BlockMeta{
-		BlockID: types.BlockID{
-			Hash: testBlockHash,
-		},
-	})
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-	res, err := cli.BlockchainInfo(context.Background(), 0, 100)
-	require.NoError(t, err)
-	require.NotNil(t, res)
-	require.Equal(t, testBlockHash, []byte(res.BlockMetas[0].BlockID.Hash))
-
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func TestValidators(t *testing.T) {
-	testHeight := int64(1)
-	testVotingPower := int64(100)
-	testValidators := types.ValidatorSet{
-		Validators: []*types.Validator{
-			{
-				VotingPower: testVotingPower,
-			},
-		},
-	}
-	stateStoreMock := &statemocks.Store{}
-	stateStoreMock.On("LoadValidators", testHeight).Return(&testValidators, nil)
-
-	blockStoreMock := &statemocks.BlockStore{}
-	blockStoreMock.On("Height").Return(testHeight)
-	blockStoreMock.On("Base").Return(int64(0))
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-
-	testPage := 1
-	testPerPage := 100
-	res, err := cli.Validators(context.Background(), &testHeight, &testPage, &testPerPage)
-	require.NoError(t, err)
-	require.NotNil(t, res)
-	require.Equal(t, testVotingPower, res.Validators[0].VotingPower)
-
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func TestBlockSearch(t *testing.T) {
-	testHeight := int64(1)
-	testBlockHash := []byte("test hash")
-	testQuery := "block.height = 1"
-	stateStoreMock := &statemocks.Store{}
-
-	blockStoreMock := &statemocks.BlockStore{}
-	eventSinkMock := &indexermocks.EventSink{}
-	eventSinkMock.On("Stop").Return(nil)
-	eventSinkMock.On("Type").Return(indexer.KV)
-	blockStoreMock.On("LoadBlock", testHeight).Return(&types.Block{
-		Header: types.Header{
-			Height: testHeight,
-		},
-	}, nil)
-	blockStoreMock.On("LoadBlockMeta", testHeight).Return(&types.BlockMeta{
-		BlockID: types.BlockID{
-			Hash: testBlockHash,
-		},
-	})
-	eventSinkMock.On("SearchBlockEvents", mock.Anything,
-		mock.MatchedBy(func(q *query.Query) bool { return testQuery == q.String() })).
-		Return([]int64{testHeight}, nil)
-	rpcConfig := config.TestRPCConfig()
-	l := log.TestingLogger()
-	d := inspect.New(rpcConfig, blockStoreMock, stateStoreMock, []indexer.EventSink{eventSinkMock}, l)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	startedWG := &sync.WaitGroup{}
-	startedWG.Add(1)
-	go func() {
-		startedWG.Done()
-		defer wg.Done()
-		require.NoError(t, d.Run(ctx))
-	}()
-	// FIXME: used to induce context switch.
-	// Determine more deterministic method for prompting a context switch
-	startedWG.Wait()
-	requireConnect(t, rpcConfig.ListenAddress, 20)
-	cli, err := httpclient.New(rpcConfig.ListenAddress)
-	require.NoError(t, err)
-
-	testPage := 1
-	testPerPage := 100
-	testOrderBy := "desc"
-	res, err := cli.BlockSearch(context.Background(), testQuery, &testPage, &testPerPage, testOrderBy)
-	require.NoError(t, err)
-	require.NotNil(t, res)
-	require.Equal(t, testBlockHash, []byte(res.Blocks[0].BlockID.Hash))
-
-	cancel()
-	wg.Wait()
-
-	blockStoreMock.AssertExpectations(t)
-	stateStoreMock.AssertExpectations(t)
-}
-
-func requireConnect(t testing.TB, addr string, retries int) {
-	parts := strings.SplitN(addr, "://", 2)
-	if len(parts) != 2 {
-		t.Fatalf("malformed address to dial: %s", addr)
-	}
-	var err error
-	for i := 0; i < retries; i++ {
-		var conn net.Conn
-		conn, err = net.Dial(parts[0], parts[1])
-		if err == nil {
-			conn.Close()
-			return
-		}
-		// FIXME attempt to yield and let the other goroutine continue execution.
-		time.Sleep(time.Microsecond * 100)
-	}
-	t.Fatalf("unable to connect to server %s after %d tries: %s", addr, retries, err)
-}

inspect/rpc/rpc.go

@@ -1,143 +0,0 @@
-package rpc
-
-import (
-	"context"
-	"net/http"
-	"time"
-
-	"github.com/rs/cors"
-
-	"github.com/tendermint/tendermint/config"
-	"github.com/tendermint/tendermint/internal/consensus"
-	"github.com/tendermint/tendermint/libs/log"
-	"github.com/tendermint/tendermint/libs/pubsub"
-	"github.com/tendermint/tendermint/rpc/core"
-	"github.com/tendermint/tendermint/rpc/jsonrpc/server"
-	"github.com/tendermint/tendermint/state"
-	"github.com/tendermint/tendermint/state/indexer"
-	"github.com/tendermint/tendermint/types"
-)
-
-// Server defines parameters for running an Inspector rpc server.
-type Server struct {
-	Addr    string // TCP address to listen on, ":http" if empty
-	Handler http.Handler
-	Logger  log.Logger
-	Config  *config.RPCConfig
-}
-
-// Routes returns the set of routes used by the Inspector server.
-//
-//nolint: lll
-func Routes(cfg config.RPCConfig, s state.Store, bs state.BlockStore, es []indexer.EventSink, logger log.Logger) core.RoutesMap {
-	env := &core.Environment{
-		Config:           cfg,
-		EventSinks:       es,
-		StateStore:       s,
-		BlockStore:       bs,
-		ConsensusReactor: waitSyncCheckerImpl{},
-		Logger:           logger,
-	}
-	return core.RoutesMap{
-		"blockchain":       server.NewRPCFunc(env.BlockchainInfo, "minHeight,maxHeight", true),
-		"consensus_params": server.NewRPCFunc(env.ConsensusParams, "height", true),
-		"block":            server.NewRPCFunc(env.Block, "height", true),
-		"block_by_hash":    server.NewRPCFunc(env.BlockByHash, "hash", true),
-		"block_results":    server.NewRPCFunc(env.BlockResults, "height", true),
-		"commit":           server.NewRPCFunc(env.Commit, "height", true),
-		"validators":       server.NewRPCFunc(env.Validators, "height,page,per_page", true),
-		"tx":               server.NewRPCFunc(env.Tx, "hash,prove", true),
-		"tx_search":        server.NewRPCFunc(env.TxSearch, "query,prove,page,per_page,order_by", false),
-		"block_search":     server.NewRPCFunc(env.BlockSearch, "query,page,per_page,order_by", false),
-	}
-}
-
-// Handler returns the http.Handler configured for use with an Inspector server. Handler
-// registers the routes on the http.Handler and also registers the websocket handler
-// and the CORS handler if specified by the configuration options.
-func Handler(rpcConfig *config.RPCConfig, routes core.RoutesMap, logger log.Logger) http.Handler {
-	mux := http.NewServeMux()
-	wmLogger := logger.With("protocol", "websocket")
-
-	var eventBus types.EventBusSubscriber
-
-	websocketDisconnectFn := func(remoteAddr string) {
-		err := eventBus.UnsubscribeAll(context.Background(), remoteAddr)
-		if err != nil && err != pubsub.ErrSubscriptionNotFound {
-			wmLogger.Error("Failed to unsubscribe addr from events", "addr", remoteAddr, "err", err)
-		}
-	}
-	wm := server.NewWebsocketManager(routes,
-		server.OnDisconnect(websocketDisconnectFn),
-		server.ReadLimit(rpcConfig.MaxBodyBytes))
-	wm.SetLogger(wmLogger)
-	mux.HandleFunc("/websocket", wm.WebsocketHandler)
-
-	server.RegisterRPCFuncs(mux, routes, logger)
-	var rootHandler http.Handler = mux
-	if rpcConfig.IsCorsEnabled() {
-		rootHandler = addCORSHandler(rpcConfig, mux)
-	}
-	return rootHandler
-}
-
-func addCORSHandler(rpcConfig *config.RPCConfig, h http.Handler) http.Handler {
-	corsMiddleware := cors.New(cors.Options{
-		AllowedOrigins: rpcConfig.CORSAllowedOrigins,
-		AllowedMethods: rpcConfig.CORSAllowedMethods,
-		AllowedHeaders: rpcConfig.CORSAllowedHeaders,
-	})
-	h = corsMiddleware.Handler(h)
-	return h
-}
-
-type waitSyncCheckerImpl struct{}
-
-func (waitSyncCheckerImpl) WaitSync() bool {
-	return false
-}
-
-func (waitSyncCheckerImpl) GetPeerState(peerID types.NodeID) (*consensus.PeerState, bool) {
-	return nil, false
-}
-
-// ListenAndServe listens on the address specified in srv.Addr and handles any
-// incoming requests over HTTP using the Inspector rpc handler specified on the server.
-func (srv *Server) ListenAndServe(ctx context.Context) error {
-	listener, err := server.Listen(srv.Addr, srv.Config.MaxOpenConnections)
-	if err != nil {
-		return err
-	}
-	go func() {
-		<-ctx.Done()
-		listener.Close()
-	}()
-	return server.Serve(listener, srv.Handler, srv.Logger, serverRPCConfig(srv.Config))
-}
-
-// ListenAndServeTLS listens on the address specified in srv.Addr. ListenAndServeTLS handles
-// incoming requests over HTTPS using the Inspector rpc handler specified on the server.
-func (srv *Server) ListenAndServeTLS(ctx context.Context, certFile, keyFile string) error {
-	listener, err := server.Listen(srv.Addr, srv.Config.MaxOpenConnections)
-	if err != nil {
-		return err
-	}
-	go func() {
-		<-ctx.Done()
-		listener.Close()
-	}()
-	return server.ServeTLS(listener, srv.Handler, certFile, keyFile, srv.Logger, serverRPCConfig(srv.Config))
-}
-
-func serverRPCConfig(r *config.RPCConfig) *server.Config {
-	cfg := server.DefaultConfig()
-	cfg.MaxBodyBytes = r.MaxBodyBytes
-	cfg.MaxHeaderBytes = r.MaxHeaderBytes
-	// If necessary adjust global WriteTimeout to ensure it's greater than
-	// TimeoutBroadcastTxCommit.
-	// See https://github.com/tendermint/tendermint/issues/3435
-	if cfg.WriteTimeout <= r.TimeoutBroadcastTxCommit {
-		cfg.WriteTimeout = r.TimeoutBroadcastTxCommit + 1*time.Second
-	}
-	return cfg
-}

libs/time/time.go

@@ -1,6 +1,7 @@
 package time
 
 import (
+	"sort"
 	"time"
 )
 
@@ -15,3 +16,43 @@ func Now() time.Time {
 func Canonical(t time.Time) time.Time {
 	return t.Round(0).UTC()
 }
+
+// WeightedTime for computing a median.
+type WeightedTime struct {
+	Time   time.Time
+	Weight int64
+}
+
+// NewWeightedTime with time and weight.
+func NewWeightedTime(time time.Time, weight int64) *WeightedTime {
+	return &WeightedTime{
+		Time:   time,
+		Weight: weight,
+	}
+}
+
+// WeightedMedian computes weighted median time for a given array of WeightedTime and the total voting power.
+func WeightedMedian(weightedTimes []*WeightedTime, totalVotingPower int64) (res time.Time) {
+	median := totalVotingPower / 2
+
+	sort.Slice(weightedTimes, func(i, j int) bool {
+		if weightedTimes[i] == nil {
+			return false
+		}
+		if weightedTimes[j] == nil {
+			return true
+		}
+		return weightedTimes[i].Time.UnixNano() < weightedTimes[j].Time.UnixNano()
+	})
+
+	for _, weightedTime := range weightedTimes {
+		if weightedTime != nil {
+			if median <= weightedTime.Weight {
+				res = weightedTime.Time
+				break
+			}
+			median -= weightedTime.Weight
+		}
+	}
+	return
+}

libs/time/time_test.go

@@ -1,55 +1,54 @@
-package state
+package time
 
 import (
 	"testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	tmtime "github.com/tendermint/tendermint/libs/time"
 )
 
 func TestWeightedMedian(t *testing.T) {
-	m := make([]*weightedTime, 3)
+	m := make([]*WeightedTime, 3)
 
-	t1 := tmtime.Now()
+	t1 := Now()
 	t2 := t1.Add(5 * time.Second)
 	t3 := t1.Add(10 * time.Second)
 
-	m[2] = newWeightedTime(t1, 33) // faulty processes
-	m[0] = newWeightedTime(t2, 40) // correct processes
-	m[1] = newWeightedTime(t3, 27) // correct processes
+	m[2] = NewWeightedTime(t1, 33) // faulty processes
+	m[0] = NewWeightedTime(t2, 40) // correct processes
+	m[1] = NewWeightedTime(t3, 27) // correct processes
 	totalVotingPower := int64(100)
 
-	median := weightedMedian(m, totalVotingPower)
+	median := WeightedMedian(m, totalVotingPower)
 	assert.Equal(t, t2, median)
 	// median always returns value between values of correct processes
 	assert.Equal(t, true, (median.After(t1) || median.Equal(t1)) &&
 		(median.Before(t3) || median.Equal(t3)))
 
-	m[1] = newWeightedTime(t1, 40) // correct processes
-	m[2] = newWeightedTime(t2, 27) // correct processes
-	m[0] = newWeightedTime(t3, 33) // faulty processes
+	m[1] = NewWeightedTime(t1, 40) // correct processes
+	m[2] = NewWeightedTime(t2, 27) // correct processes
+	m[0] = NewWeightedTime(t3, 33) // faulty processes
 	totalVotingPower = int64(100)
 
-	median = weightedMedian(m, totalVotingPower)
+	median = WeightedMedian(m, totalVotingPower)
 	assert.Equal(t, t2, median)
 	// median always returns value between values of correct processes
 	assert.Equal(t, true, (median.After(t1) || median.Equal(t1)) &&
 		(median.Before(t2) || median.Equal(t2)))
 
-	m = make([]*weightedTime, 8)
+	m = make([]*WeightedTime, 8)
 	t4 := t1.Add(15 * time.Second)
 	t5 := t1.Add(60 * time.Second)
 
-	m[3] = newWeightedTime(t1, 10) // correct processes
-	m[1] = newWeightedTime(t2, 10) // correct processes
-	m[5] = newWeightedTime(t2, 10) // correct processes
-	m[4] = newWeightedTime(t3, 23) // faulty processes
-	m[0] = newWeightedTime(t4, 20) // correct processes
-	m[7] = newWeightedTime(t5, 10) // faulty processes
+	m[3] = NewWeightedTime(t1, 10) // correct processes
+	m[1] = NewWeightedTime(t2, 10) // correct processes
+	m[5] = NewWeightedTime(t2, 10) // correct processes
+	m[4] = NewWeightedTime(t3, 23) // faulty processes
+	m[0] = NewWeightedTime(t4, 20) // correct processes
+	m[7] = NewWeightedTime(t5, 10) // faulty processes
 	totalVotingPower = int64(83)
 
-	median = weightedMedian(m, totalVotingPower)
+	median = WeightedMedian(m, totalVotingPower)
 	assert.Equal(t, t3, median)
 	// median always returns value between values of correct processes
 	assert.Equal(t, true, (median.After(t1) || median.Equal(t1)) &&

light/proxy/proxy.go

@@ -113,7 +113,7 @@ func (p *Proxy) listen() (net.Listener, *http.ServeMux, error) {
 	}
 
 	// 4) Start listening for new connections.
-	listener, err := rpcserver.Listen(p.Addr, p.Config.MaxOpenConnections)
+	listener, err := rpcserver.Listen(p.Addr, p.Config)
 	if err != nil {
 		return nil, mux, err
 	}

node/node.go

@@ -441,23 +441,22 @@ func makeNode(config *cfg.Config,
 			ProxyAppQuery:   proxyApp.Query(),
 			ProxyAppMempool: proxyApp.Mempool(),
 
-			StateStore:     stateStore,
-			BlockStore:     blockStore,
-			EvidencePool:   evPool,
-			ConsensusState: csState,
-
-			ConsensusReactor: csReactor,
+			StateStore:       stateStore,
+			BlockStore:       blockStore,
+			EvidencePool:     evPool,
+			ConsensusState:   csState,
 			BlockSyncReactor: bcReactor.(cs.BlockSyncReactor),
 
 			P2PPeers:    sw,
 			PeerManager: peerManager,
 
-			GenDoc:     genDoc,
-			EventSinks: eventSinks,
-			EventBus:   eventBus,
-			Mempool:    mp,
-			Logger:     logger.With("module", "rpc"),
-			Config:     *config.RPC,
+			GenDoc:           genDoc,
+			EventSinks:       eventSinks,
+			ConsensusReactor: csReactor,
+			EventBus:         eventBus,
+			Mempool:          mp,
+			Logger:           logger.With("module", "rpc"),
+			Config:           *config.RPC,
 		},
 	}
 
@@ -826,7 +825,7 @@ func (n *nodeImpl) startRPC() ([]net.Listener, error) {
 		rpcserver.RegisterRPCFuncs(mux, routes, rpcLogger)
 		listener, err := rpcserver.Listen(
 			listenAddr,
-			config.MaxOpenConnections,
+			config,
 		)
 		if err != nil {
 			return nil, err
@@ -884,7 +883,7 @@ func (n *nodeImpl) startRPC() ([]net.Listener, error) {
 		if config.WriteTimeout <= n.config.RPC.TimeoutBroadcastTxCommit {
 			config.WriteTimeout = n.config.RPC.TimeoutBroadcastTxCommit + 1*time.Second
 		}
-		listener, err := rpcserver.Listen(grpcListenAddr, config.MaxOpenConnections)
+		listener, err := rpcserver.Listen(grpcListenAddr, config)
 		if err != nil {
 			return nil, err
 		}

node/setup.go

@@ -8,6 +8,7 @@ import (
 	"math"
 	"net"
 	_ "net/http/pprof" // nolint: gosec // securely exposed on separate, optional port
+	"strings"
 	"time"
 
 	dbm "github.com/tendermint/tm-db"
@@ -32,7 +33,9 @@ import (
 	"github.com/tendermint/tendermint/proxy"
 	sm "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/kv"
+	null "github.com/tendermint/tendermint/state/indexer/sink/null"
+	psql "github.com/tendermint/tendermint/state/indexer/sink/psql"
 	"github.com/tendermint/tendermint/store"
 	"github.com/tendermint/tendermint/types"
 	"github.com/tendermint/tendermint/version"
@@ -75,9 +78,56 @@ func createAndStartIndexerService(
 	logger log.Logger,
 	chainID string,
 ) (*indexer.Service, []indexer.EventSink, error) {
-	eventSinks, err := sink.EventSinksFromConfig(config, dbProvider, chainID)
-	if err != nil {
-		return nil, nil, err
+
+	eventSinks := []indexer.EventSink{}
+
+	// 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
+	}
+
+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.
+			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")
+		}
+	}
+
+	if len(eventSinks) == 0 {
+		eventSinks = []indexer.EventSink{null.NewEventSink()}
 	}
 
 	indexerService := indexer.NewIndexerService(eventSinks, eventBus)

release_notes.md

@@ -0,0 +1 @@
+<!-- TODO: add in release notes prior to release. -->

rpc/core/env.go

@@ -58,11 +58,6 @@ type peers interface {
 	Peers() p2p.IPeerSet
 }
 
-type consensusReactor interface {
-	WaitSync() bool
-	GetPeerState(peerID types.NodeID) (*consensus.PeerState, bool)
-}
-
 type peerManager interface {
 	Peers() []types.NodeID
 	Addresses(types.NodeID) []p2p.NodeAddress
@@ -77,12 +72,11 @@ type Environment struct {
 	ProxyAppMempool proxy.AppConnMempool
 
 	// interfaces defined in types and above
-	StateStore       sm.Store
-	BlockStore       sm.BlockStore
-	EvidencePool     sm.EvidencePool
-	ConsensusState   consensusState
-	ConsensusReactor consensusReactor
-	P2PPeers         peers
+	StateStore     sm.Store
+	BlockStore     sm.BlockStore
+	EvidencePool   sm.EvidencePool
+	ConsensusState consensusState
+	P2PPeers       peers
 
 	// Legacy p2p stack
 	P2PTransport transport
@@ -94,6 +88,7 @@ type Environment struct {
 	PubKey           crypto.PubKey
 	GenDoc           *types.GenesisDoc // cache the genesis structure
 	EventSinks       []indexer.EventSink
+	ConsensusReactor *consensus.Reactor
 	EventBus         *types.EventBus // thread safe
 	Mempool          mempl.Mempool
 	BlockSyncReactor consensus.BlockSyncReactor

rpc/jsonrpc/jsonrpc_test.go

@@ -110,7 +110,7 @@ func setup() {
 	wm.SetLogger(tcpLogger)
 	mux.HandleFunc(websocketEndpoint, wm.WebsocketHandler)
 	config := server.DefaultConfig()
-	listener1, err := server.Listen(tcpAddr, config.MaxOpenConnections)
+	listener1, err := server.Listen(tcpAddr, config)
 	if err != nil {
 		panic(err)
 	}
@@ -126,7 +126,7 @@ func setup() {
 	wm = server.NewWebsocketManager(Routes)
 	wm.SetLogger(unixLogger)
 	mux2.HandleFunc(websocketEndpoint, wm.WebsocketHandler)
-	listener2, err := server.Listen(unixAddr, config.MaxOpenConnections)
+	listener2, err := server.Listen(unixAddr, config)
 	if err != nil {
 		panic(err)
 	}

rpc/jsonrpc/server/http_server.go

@@ -261,7 +261,7 @@ func (h maxBytesHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 // Listen starts a new net.Listener on the given address.
 // It returns an error if the address is invalid or the call to Listen() fails.
-func Listen(addr string, maxOpenConnections int) (listener net.Listener, err error) {
+func Listen(addr string, config *Config) (listener net.Listener, err error) {
 	parts := strings.SplitN(addr, "://", 2)
 	if len(parts) != 2 {
 		return nil, fmt.Errorf(
@@ -274,8 +274,8 @@ func Listen(addr string, maxOpenConnections int) (listener net.Listener, err err
 	if err != nil {
 		return nil, fmt.Errorf("failed to listen on %v: %v", addr, err)
 	}
-	if maxOpenConnections > 0 {
-		listener = netutil.LimitListener(listener, maxOpenConnections)
+	if config.MaxOpenConnections > 0 {
+		listener = netutil.LimitListener(listener, config.MaxOpenConnections)
 	}
 
 	return listener, nil

rpc/jsonrpc/server/http_server_test.go

@@ -39,7 +39,8 @@ func TestMaxOpenConnections(t *testing.T) {
 		fmt.Fprint(w, "some body")
 	})
 	config := DefaultConfig()
-	l, err := Listen("tcp://127.0.0.1:0", max)
+	config.MaxOpenConnections = max
+	l, err := Listen("tcp://127.0.0.1:0", config)
 	require.NoError(t, err)
 	defer l.Close()
 	go Serve(l, mux, log.TestingLogger(), config) //nolint:errcheck // ignore for tests

rpc/jsonrpc/test/main.go

@@ -33,7 +33,7 @@ func main() {
 
 	rpcserver.RegisterRPCFuncs(mux, routes, logger)
 	config := rpcserver.DefaultConfig()
-	listener, err := rpcserver.Listen("tcp://127.0.0.1:8008", config.MaxOpenConnections)
+	listener, err := rpcserver.Listen("tcp://127.0.0.1:8008", config)
 	if err != nil {
 		tmos.Exit(err.Error())
 	}

state/indexer/indexer_service.go

@@ -51,47 +51,43 @@ func (is *Service) OnStart() error {
 
 	go func() {
 		for {
-			select {
-			case <-blockHeadersSub.Canceled():
-				return
-			case msg := <-blockHeadersSub.Out():
+			msg := <-blockHeadersSub.Out()
 
-				eventDataHeader := msg.Data().(types.EventDataNewBlockHeader)
-				height := eventDataHeader.Header.Height
-				batch := NewBatch(eventDataHeader.NumTxs)
+			eventDataHeader := msg.Data().(types.EventDataNewBlockHeader)
+			height := eventDataHeader.Header.Height
+			batch := NewBatch(eventDataHeader.NumTxs)
 
-				for i := int64(0); i < eventDataHeader.NumTxs; i++ {
-					msg2 := <-txsSub.Out()
-					txResult := msg2.Data().(types.EventDataTx).TxResult
+			for i := int64(0); i < eventDataHeader.NumTxs; i++ {
+				msg2 := <-txsSub.Out()
+				txResult := msg2.Data().(types.EventDataTx).TxResult
 
-					if err = batch.Add(&txResult); err != nil {
-						is.Logger.Error(
-							"failed to add tx to batch",
-							"height", height,
-							"index", txResult.Index,
-							"err", err,
-						)
-					}
+				if err = batch.Add(&txResult); err != nil {
+					is.Logger.Error(
+						"failed to add tx to batch",
+						"height", height,
+						"index", txResult.Index,
+						"err", err,
+					)
+				}
+			}
+
+			if !IndexingEnabled(is.eventSinks) {
+				continue
+			}
+
+			for _, sink := range is.eventSinks {
+				if err := sink.IndexBlockEvents(eventDataHeader); err != nil {
+					is.Logger.Error("failed to index block", "height", height, "err", err)
+				} else {
+					is.Logger.Debug("indexed block", "height", height, "sink", sink.Type())
 				}
 
-				if !IndexingEnabled(is.eventSinks) {
-					continue
-				}
-
-				for _, sink := range is.eventSinks {
-					if err := sink.IndexBlockEvents(eventDataHeader); err != nil {
-						is.Logger.Error("failed to index block", "height", height, "err", err)
+				if len(batch.Ops) > 0 {
+					err := sink.IndexTxEvents(batch.Ops)
+					if err != nil {
+						is.Logger.Error("failed to index block txs", "height", height, "err", err)
 					} else {
-						is.Logger.Debug("indexed block", "height", height, "sink", sink.Type())
-					}
-
-					if len(batch.Ops) > 0 {
-						err := sink.IndexTxEvents(batch.Ops)
-						if err != nil {
-							is.Logger.Error("failed to index block txs", "height", height, "err", err)
-						} else {
-							is.Logger.Debug("indexed txs", "height", height, "sink", sink.Type())
-						}
+						is.Logger.Debug("indexed txs", "height", height, "sink", sink.Type())
 					}
 				}
 			}

state/indexer/sink/sink.go

@@ -1,65 +0,0 @@
-package sink
-
-import (
-	"errors"
-	"strings"
-
-	"github.com/tendermint/tendermint/config"
-	"github.com/tendermint/tendermint/state/indexer"
-	"github.com/tendermint/tendermint/state/indexer/sink/kv"
-	"github.com/tendermint/tendermint/state/indexer/sink/null"
-	"github.com/tendermint/tendermint/state/indexer/sink/psql"
-)
-
-// EventSinksFromConfig constructs a slice of indexer.EventSink using the provided
-// configuration.
-//
-//nolint:lll
-func EventSinksFromConfig(cfg *config.Config, dbProvider config.DBProvider, chainID string) ([]indexer.EventSink, error) {
-	if len(cfg.TxIndex.Indexer) == 0 {
-		return []indexer.EventSink{null.NewEventSink()}, nil
-	}
-
-	// check for duplicated sinks
-	sinks := map[string]struct{}{}
-	for _, s := range cfg.TxIndex.Indexer {
-		sl := strings.ToLower(s)
-		if _, ok := sinks[sl]; ok {
-			return nil, errors.New("found duplicated sinks, please check the tx-index section in the config.toml")
-		}
-		sinks[sl] = struct{}{}
-	}
-	eventSinks := []indexer.EventSink{}
-	for k := range sinks {
-		switch indexer.EventSinkType(k) {
-		case indexer.NULL:
-			// When we see null in the config, the eventsinks will be reset with the
-			// nullEventSink.
-			return []indexer.EventSink{null.NewEventSink()}, nil
-
-		case indexer.KV:
-			store, err := dbProvider(&config.DBContext{ID: "tx_index", Config: cfg})
-			if err != nil {
-				return nil, err
-			}
-
-			eventSinks = append(eventSinks, kv.NewEventSink(store))
-
-		case 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")
-		}
-	}
-	return eventSinks, nil
-
-}

state/state.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/gogo/protobuf/proto"
 
+	tmtime "github.com/tendermint/tendermint/libs/time"
 	tmstate "github.com/tendermint/tendermint/proto/tendermint/state"
 	tmversion "github.com/tendermint/tendermint/proto/tendermint/version"
 	"github.com/tendermint/tendermint/types"
@@ -286,7 +287,7 @@ func (state State) MakeBlock(
 // the votes sent by honest processes, i.e., a faulty processes can not arbitrarily increase or decrease the
 // computed value.
 func MedianTime(commit *types.Commit, validators *types.ValidatorSet) time.Time {
-	weightedTimes := make([]*weightedTime, len(commit.Signatures))
+	weightedTimes := make([]*tmtime.WeightedTime, len(commit.Signatures))
 	totalVotingPower := int64(0)
 
 	for i, commitSig := range commit.Signatures {
@@ -297,11 +298,11 @@ func MedianTime(commit *types.Commit, validators *types.ValidatorSet) time.Time
 		// If there's no condition, TestValidateBlockCommit panics; not needed normally.
 		if validator != nil {
 			totalVotingPower += validator.VotingPower
-			weightedTimes[i] = newWeightedTime(commitSig.Timestamp, validator.VotingPower)
+			weightedTimes[i] = tmtime.NewWeightedTime(commitSig.Timestamp, validator.VotingPower)
 		}
 	}
 
-	return weightedMedian(weightedTimes, totalVotingPower)
+	return tmtime.WeightedMedian(weightedTimes, totalVotingPower)
 }
 
 //------------------------------------------------------------------------

state/time.go

@@ -1,46 +0,0 @@
-package state
-
-import (
-	"sort"
-	"time"
-)
-
-// weightedTime for computing a median.
-type weightedTime struct {
-	Time   time.Time
-	Weight int64
-}
-
-// newWeightedTime with time and weight.
-func newWeightedTime(time time.Time, weight int64) *weightedTime {
-	return &weightedTime{
-		Time:   time,
-		Weight: weight,
-	}
-}
-
-// weightedMedian computes weighted median time for a given array of WeightedTime and the total voting power.
-func weightedMedian(weightedTimes []*weightedTime, totalVotingPower int64) (res time.Time) {
-	median := totalVotingPower / 2
-
-	sort.Slice(weightedTimes, func(i, j int) bool {
-		if weightedTimes[i] == nil {
-			return false
-		}
-		if weightedTimes[j] == nil {
-			return true
-		}
-		return weightedTimes[i].Time.UnixNano() < weightedTimes[j].Time.UnixNano()
-	})
-
-	for _, weightedTime := range weightedTimes {
-		if weightedTime != nil {
-			if median <= weightedTime.Weight {
-				res = weightedTime.Time
-				break
-			}
-			median -= weightedTime.Weight
-		}
-	}
-	return
-}

test/e2e/pkg/testnet.go

@@ -417,16 +417,6 @@ func (t Testnet) ArchiveNodes() []*Node {
 	return nodes
 }
 
-// RandomNode returns a random non-seed node.
-func (t Testnet) RandomNode() *Node {
-	for {
-		node := t.Nodes[rand.Intn(len(t.Nodes))]
-		if node.Mode != ModeSeed {
-			return node
-		}
-	}
-}
-
 // IPv6 returns true if the testnet is an IPv6 network.
 func (t Testnet) IPv6() bool {
 	return t.IP.IP.To4() == nil

test/e2e/runner/load.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"container/ring"
 	"context"
 	"crypto/rand"
 	"errors"
@@ -95,32 +96,56 @@ func loadGenerate(ctx context.Context, chTx chan<- types.Tx, multiplier int, siz
 func loadProcess(ctx context.Context, testnet *e2e.Testnet, chTx <-chan types.Tx, chSuccess chan<- types.Tx) {
 	// Each worker gets its own client to each node, which allows for some
 	// concurrency while still bounding it.
-	clients := map[string]*rpchttp.HTTP{}
 
-	var err error
-	for tx := range chTx {
-		node := testnet.RandomNode()
+	clients := make([]*rpchttp.HTTP, 0, len(testnet.Nodes))
 
-		client, ok := clients[node.Name]
-		if !ok {
-			client, err = node.Client()
-			if err != nil {
-				continue
-			}
-
-			// check that the node is up
-			_, err = client.Health(ctx)
-			if err != nil {
-				continue
-			}
-
-			clients[node.Name] = client
+	for idx := range testnet.Nodes {
+		if testnet.Nodes[idx].Mode == e2e.ModeSeed {
+			continue
 		}
-
-		if _, err = client.BroadcastTxSync(ctx, tx); err != nil {
+		client, err := testnet.Nodes[idx].Client()
+		if err != nil {
 			continue
 		}
 
-		chSuccess <- tx
+		clients = append(clients, client)
+	}
+
+	if len(clients) == 0 {
+		panic("no clients to process load")
+	}
+
+	clientRing := ring.New(len(clients))
+	for idx := range clients {
+		clientRing.Value = clients[idx]
+		clientRing = clientRing.Next()
+	}
+
+	var err error
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case tx := <-chTx:
+			clientRing = clientRing.Next()
+			client := clientRing.Value.(*rpchttp.HTTP)
+
+			if _, err := client.Health(ctx); err != nil {
+				continue
+			}
+
+			if _, err = client.BroadcastTxSync(ctx, tx); err != nil {
+				continue
+			}
+
+			select {
+			case chSuccess <- tx:
+				continue
+			case <-ctx.Done():
+				return
+			}
+
+		}
 	}
 }