Merge branch 'master' into wb/pbts-rebase-master

internal/consensus/byzantine_test.go

@@ -256,7 +256,6 @@ func TestByzantinePrevoteEquivocation(t *testing.T) {
 				}
 
 				msg, err := s.Next(ctx)
-
 				assert.NoError(t, err)
 				if err != nil {
 					cancel()

internal/consensus/metrics.go

@@ -64,6 +64,22 @@ type Metrics struct {
 
 	// Histogram of time taken per step annotated with reason that the step proceeded.
 	StepTime metrics.Histogram
+
+	// QuroumPrevoteMessageDelay is the interval in seconds between the proposal
+	// timestamp and the timestamp of the earliest prevote that achieved a quorum
+	// during the prevote step.
+	//
+	// To compute it, sum the voting power over each prevote received, in increasing
+	// order of timestamp. The timestamp of the first prevote to increase the sum to
+	// be above 2/3 of the total voting power of the network defines the endpoint
+	// the endpoint of the interval. Subtract the proposal timestamp from this endpoint
+	// to obtain the quorum delay.
+	QuorumPrevoteMessageDelay metrics.Gauge
+
+	// FullPrevoteMessageDelay is the interval in seconds between the proposal
+	// timestamp and the timestamp of the latest prevote in a round where 100%
+	// of the voting power on the network issued prevotes.
+	FullPrevoteMessageDelay metrics.Gauge
 }
 
 // PrometheusMetrics returns Metrics build using Prometheus client library.
@@ -196,6 +212,20 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Name:      "step_time",
 			Help:      "Time spent per step.",
 		}, append(labels, "step", "reason")).With(labelsAndValues...),
+		QuorumPrevoteMessageDelay: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "quorum_prevote_message_delay",
+			Help: "Difference in seconds between the proposal timestamp and the timestamp " +
+				"of the latest prevote that achieved a quorum in the prevote step.",
+		}, labels).With(labelsAndValues...),
+		FullPrevoteMessageDelay: prometheus.NewGaugeFrom(stdprometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "full_prevote_message_delay",
+			Help: "Difference in seconds between the proposal timestamp and the timestamp " +
+				"of the latest prevote that achieved 100% of the voting power in the prevote step.",
+		}, labels).With(labelsAndValues...),
 	}
 }
 
@@ -219,13 +249,15 @@ func NopMetrics() *Metrics {
 
 		BlockIntervalSeconds: discard.NewHistogram(),
 
-		NumTxs:          discard.NewGauge(),
-		BlockSizeBytes:  discard.NewHistogram(),
-		TotalTxs:        discard.NewGauge(),
-		CommittedHeight: discard.NewGauge(),
-		BlockSyncing:    discard.NewGauge(),
-		StateSyncing:    discard.NewGauge(),
-		BlockParts:      discard.NewCounter(),
+		NumTxs:                    discard.NewGauge(),
+		BlockSizeBytes:            discard.NewHistogram(),
+		TotalTxs:                  discard.NewGauge(),
+		CommittedHeight:           discard.NewGauge(),
+		BlockSyncing:              discard.NewGauge(),
+		StateSyncing:              discard.NewGauge(),
+		BlockParts:                discard.NewCounter(),
+		QuorumPrevoteMessageDelay: discard.NewGauge(),
+		FullPrevoteMessageDelay:   discard.NewGauge(),
 	}
 }
 

internal/consensus/replay_test.go

@@ -201,6 +201,8 @@ LOOP:
 		i++
 
 		select {
+		case <-rctx.Done():
+			t.Fatal("context canceled before test completed")
 		case err := <-walPanicked:
 			// make sure we can make blocks after a crash
 			startNewStateAndWaitForBlock(ctx, t, consensusReplayConfig, cs.Height, blockDB, stateStore)

internal/consensus/state.go

@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"runtime/debug"
+	"sort"
 	"sync"
 	"time"
 
@@ -1764,6 +1765,8 @@ func (cs *State) finalizeCommit(ctx context.Context, height int64) {
 		return
 	}
 
+	cs.calculatePrevoteMessageDelayMetrics()
+
 	blockID, ok := cs.Votes.Precommits(cs.CommitRound).TwoThirdsMajority()
 	block, blockParts := cs.ProposalBlock, cs.ProposalBlockParts
 
@@ -2447,6 +2450,26 @@ func (cs *State) checkDoubleSigningRisk(height int64) error {
 	return nil
 }
 
+func (cs *State) calculatePrevoteMessageDelayMetrics() {
+	ps := cs.Votes.Prevotes(cs.Round)
+	pl := ps.List()
+	sort.Slice(pl, func(i, j int) bool {
+		return pl[i].Timestamp.Before(pl[j].Timestamp)
+	})
+	var votingPowerSeen int64
+	for _, v := range pl {
+		_, val := cs.Validators.GetByAddress(v.ValidatorAddress)
+		votingPowerSeen += val.VotingPower
+		if votingPowerSeen >= cs.Validators.TotalVotingPower()*2/3+1 {
+			cs.metrics.QuorumPrevoteMessageDelay.Set(v.Timestamp.Sub(cs.Proposal.Timestamp).Seconds())
+			break
+		}
+	}
+	if ps.HasAll() {
+		cs.metrics.FullPrevoteMessageDelay.Set(pl[len(pl)-1].Timestamp.Sub(cs.Proposal.Timestamp).Seconds())
+	}
+}
+
 //---------------------------------------------------------
 
 func CompareHRS(h1 int64, r1 int32, s1 cstypes.RoundStepType, h2 int64, r2 int32, s2 cstypes.RoundStepType) int {

internal/consensus/state_test.go

@@ -2762,7 +2762,11 @@ func subscribe(
 				t.Errorf("Subscription for %v unexpectedly terminated: %v", q, err)
 				return
 			}
-			ch <- next
+			select {
+			case ch <- next:
+			case <-ctx.Done():
+				return
+			}
 		}
 	}()
 	return ch

internal/consensus/wal_test.go

@@ -3,6 +3,7 @@ package consensus
 import (
 	"bytes"
 	"context"
+	"errors"
 	"path/filepath"
 
 	"testing"
@@ -15,6 +16,7 @@ import (
 	"github.com/tendermint/tendermint/internal/consensus/types"
 	"github.com/tendermint/tendermint/internal/libs/autofile"
 	"github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/libs/service"
 	tmtime "github.com/tendermint/tendermint/libs/time"
 	tmtypes "github.com/tendermint/tendermint/types"
 )
@@ -185,7 +187,9 @@ func TestWALPeriodicSync(t *testing.T) {
 	require.NoError(t, wal.Start(ctx))
 	t.Cleanup(func() {
 		if err := wal.Stop(); err != nil {
-			t.Error(err)
+			if !errors.Is(err, service.ErrAlreadyStopped) {
+				t.Error(err)
+			}
 		}
 		wal.Wait()
 	})

internal/jsontypes/jsontypes.go

@@ -0,0 +1,109 @@
+// Package jsontypes supports decoding for interface types whose concrete
+// implementations need to be stored as JSON. To do this, concrete values are
+// packaged in wrapper objects having the form:
+//
+//   {
+//     "type": "<type-tag>",
+//     "value": <json-encoding-of-value>
+//   }
+//
+// This package provides a registry for type tag strings and functions to
+// encode and decode wrapper objects.
+package jsontypes
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"reflect"
+)
+
+// The Tagged interface must be implemented by a type in order to register it
+// with the jsontypes package. The TypeTag method returns a string label that
+// is used to distinguish objects of that type.
+type Tagged interface {
+	TypeTag() string
+}
+
+// registry records the mapping from type tags to value types.  Values in this
+// map must be normalized to non-pointer types.
+var registry = struct {
+	types map[string]reflect.Type
+}{types: make(map[string]reflect.Type)}
+
+// register adds v to the type registry. It reports an error if the tag
+// returned by v is already registered.
+func register(v Tagged) error {
+	tag := v.TypeTag()
+	if t, ok := registry.types[tag]; ok {
+		return fmt.Errorf("type tag %q already registered to %v", tag, t)
+	}
+	typ := reflect.TypeOf(v)
+	if typ.Kind() == reflect.Ptr {
+		typ = typ.Elem()
+	}
+	registry.types[tag] = typ
+	return nil
+}
+
+// MustRegister adds v to the type registry. It will panic if the tag returned
+// by v is already registered. This function is meant for use during program
+// initialization.
+func MustRegister(v Tagged) {
+	if err := register(v); err != nil {
+		panic(err)
+	}
+}
+
+type wrapper struct {
+	Type  string          `json:"type"`
+	Value json.RawMessage `json:"value"`
+}
+
+// Marshal marshals a JSON wrapper object containing v. If v == nil, Marshal
+// returns the JSON "null" value without error.
+func Marshal(v Tagged) ([]byte, error) {
+	if v == nil {
+		return []byte("null"), nil
+	}
+	data, err := json.Marshal(v)
+	if err != nil {
+		return nil, err
+	}
+	return json.Marshal(wrapper{
+		Type:  v.TypeTag(),
+		Value: data,
+	})
+}
+
+// Unmarshal unmarshals a JSON wrapper object into v. It reports an error if
+// the data do not encode a valid wrapper object, if the wrapper's type tag is
+// not registered with jsontypes, or if the resulting value is not compatible
+// with the type of v.
+func Unmarshal(data []byte, v interface{}) error {
+	// Verify that the target is some kind of pointer.
+	target := reflect.ValueOf(v)
+	if target.Kind() != reflect.Ptr {
+		return fmt.Errorf("target %T is not a pointer", v)
+	}
+
+	var w wrapper
+	dec := json.NewDecoder(bytes.NewReader(data))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&w); err != nil {
+		return fmt.Errorf("invalid type wrapper: %w", err)
+	}
+	typ, ok := registry.types[w.Type]
+	if !ok {
+		return fmt.Errorf("unknown type tag: %q", w.Type)
+	} else if !typ.AssignableTo(target.Elem().Type()) {
+		return fmt.Errorf("type %v not assignable to %T", typ, v)
+	}
+
+	obj := reflect.New(typ)
+	if err := json.Unmarshal(w.Value, obj.Interface()); err != nil {
+		return fmt.Errorf("decoding wrapped value: %w", err)
+	}
+	target.Elem().Set(obj.Elem())
+	return nil
+}

internal/jsontypes/jsontypes_test.go

@@ -0,0 +1,83 @@
+package jsontypes_test
+
+import (
+	"testing"
+
+	"github.com/tendermint/tendermint/internal/jsontypes"
+)
+
+type testType struct {
+	Field string `json:"field"`
+}
+
+func (*testType) TypeTag() string { return "test/TaggedType" }
+
+func TestRoundTrip(t *testing.T) {
+	const wantEncoded = `{"type":"test/TaggedType","value":{"field":"hello"}}`
+
+	t.Run("MustRegisterOK", func(t *testing.T) {
+		defer func() {
+			if x := recover(); x != nil {
+				t.Fatalf("Registration panicked: %v", x)
+			}
+		}()
+		jsontypes.MustRegister((*testType)(nil))
+	})
+
+	t.Run("MustRegisterFail", func(t *testing.T) {
+		defer func() {
+			if x := recover(); x != nil {
+				t.Logf("Got expected panic: %v", x)
+			}
+		}()
+		jsontypes.MustRegister((*testType)(nil))
+		t.Fatal("Registration should not have succeeded")
+	})
+
+	t.Run("MarshalNil", func(t *testing.T) {
+		bits, err := jsontypes.Marshal(nil)
+		if err != nil {
+			t.Fatalf("Marshal failed: %v", err)
+		}
+		if got := string(bits); got != "null" {
+			t.Errorf("Marshal nil: got %#q, want null", got)
+		}
+	})
+
+	t.Run("RoundTrip", func(t *testing.T) {
+		obj := testType{Field: "hello"}
+		bits, err := jsontypes.Marshal(&obj)
+		if err != nil {
+			t.Fatalf("Marshal %T failed: %v", obj, err)
+		}
+		if got := string(bits); got != wantEncoded {
+			t.Errorf("Marshal %T: got %#q, want %#q", obj, got, wantEncoded)
+		}
+
+		var cmp testType
+		if err := jsontypes.Unmarshal(bits, &cmp); err != nil {
+			t.Errorf("Unmarshal %#q failed: %v", string(bits), err)
+		}
+		if obj != cmp {
+			t.Errorf("Unmarshal %#q: got %+v, want %+v", string(bits), cmp, obj)
+		}
+	})
+
+	t.Run("Unregistered", func(t *testing.T) {
+		obj := testType{Field: "hello"}
+		bits, err := jsontypes.Marshal(&obj)
+		if err != nil {
+			t.Fatalf("Marshal %T failed: %v", obj, err)
+		}
+		if got := string(bits); got != wantEncoded {
+			t.Errorf("Marshal %T: got %#q, want %#q", obj, got, wantEncoded)
+		}
+
+		var cmp struct {
+			Field string `json:"field"`
+		}
+		if err := jsontypes.Unmarshal(bits, &cmp); err != nil {
+			t.Errorf("Unmarshal %#q: got %+v, want %+v", string(bits), cmp, obj)
+		}
+	})
+}

internal/libs/protoio/io_test.go

@@ -95,7 +95,7 @@ func iotest(t *testing.T, writer protoio.WriteCloser, reader protoio.ReadCloser)
 		}
 		i++
 	}
-	require.Equal(t, size, i)
+	require.Equal(t, size, i, "messages read ≠ messages written")
 	if err := reader.Close(); err != nil {
 		return err
 	}

internal/p2p/conn/secret_connection_test.go

@@ -52,6 +52,7 @@ func (pk privKeyWithNilPubKey) Sign(msg []byte) ([]byte, error) { return pk.orig
 func (pk privKeyWithNilPubKey) PubKey() crypto.PubKey           { return nil }
 func (pk privKeyWithNilPubKey) Equals(pk2 crypto.PrivKey) bool  { return pk.orig.Equals(pk2) }
 func (pk privKeyWithNilPubKey) Type() string                    { return "privKeyWithNilPubKey" }
+func (privKeyWithNilPubKey) TypeTag() string                    { return "test/privKeyWithNilPubKey" }
 
 func TestSecretConnectionHandshake(t *testing.T) {
 	fooSecConn, barSecConn := makeSecretConnPair(t)

internal/p2p/pex/reactor.go

@@ -158,10 +158,13 @@ func (r *Reactor) OnStop() {}
 func (r *Reactor) processPexCh(ctx context.Context) {
 	timer := time.NewTimer(0)
 	defer timer.Stop()
+
+	r.mtx.Lock()
 	var (
 		duration = r.calculateNextRequestTime()
 		err      error
 	)
+	r.mtx.Unlock()
 
 	incoming := make(chan *p2p.Envelope)
 	go func() {
@@ -191,7 +194,10 @@ func (r *Reactor) processPexCh(ctx context.Context) {
 			}
 		// inbound requests for new peers or responses to requests sent by this
 		// reactor
-		case envelope := <-incoming:
+		case envelope, ok := <-incoming:
+			if !ok {
+				return
+			}
 			duration, err = r.handleMessage(ctx, r.pexCh.ID, envelope)
 			if err != nil {
 				r.logger.Error("failed to process message", "ch_id", r.pexCh.ID, "envelope", envelope, "err", err)
@@ -377,7 +383,8 @@ func (r *Reactor) sendRequestForPeers(ctx context.Context) (time.Duration, error
 // as possible. As the node becomes more familiar with the network the ratio of
 // new nodes will plummet to a very small number, meaning the interval expands
 // to its upper bound.
-// CONTRACT: Must use a write lock as nextRequestTime is updated
+//
+// CONTRACT: The caller must hold r.mtx exclusively when calling this method.
 func (r *Reactor) calculateNextRequestTime() time.Duration {
 	// check if the peer store is full. If so then there is no need
 	// to send peer requests too often

internal/rpc/core/mempool.go

@@ -10,6 +10,7 @@ import (
 	abci "github.com/tendermint/tendermint/abci/types"
 	"github.com/tendermint/tendermint/internal/mempool"
 	"github.com/tendermint/tendermint/internal/state/indexer"
+	tmmath "github.com/tendermint/tendermint/libs/math"
 	"github.com/tendermint/tendermint/rpc/coretypes"
 	"github.com/tendermint/tendermint/types"
 )
@@ -117,19 +118,26 @@ func (env *Environment) BroadcastTxCommit(ctx context.Context, tx types.Tx) (*co
 	}
 }
 
-// UnconfirmedTxs gets unconfirmed transactions (maximum ?limit entries)
-// including their number.
+// UnconfirmedTxs gets unconfirmed transactions from the mempool in order of priority
 // More: https://docs.tendermint.com/master/rpc/#/Info/unconfirmed_txs
-func (env *Environment) UnconfirmedTxs(ctx context.Context, limitPtr *int) (*coretypes.ResultUnconfirmedTxs, error) {
-	// reuse per_page validator
-	limit := env.validatePerPage(limitPtr)
+func (env *Environment) UnconfirmedTxs(ctx context.Context, pagePtr, perPagePtr *int) (*coretypes.ResultUnconfirmedTxs, error) {
+	totalCount := env.Mempool.Size()
+	perPage := env.validatePerPage(perPagePtr)
+	page, err := validatePage(pagePtr, perPage, totalCount)
+	if err != nil {
+		return nil, err
+	}
+
+	skipCount := validateSkipCount(page, perPage)
+
+	txs := env.Mempool.ReapMaxTxs(skipCount + tmmath.MinInt(perPage, totalCount-skipCount))
+	result := txs[skipCount:]
 
-	txs := env.Mempool.ReapMaxTxs(limit)
 	return &coretypes.ResultUnconfirmedTxs{
-		Count:      len(txs),
-		Total:      env.Mempool.Size(),
+		Count:      len(result),
+		Total:      totalCount,
 		TotalBytes: env.Mempool.SizeBytes(),
-		Txs:        txs}, nil
+		Txs:        result}, nil
 }
 
 // NumUnconfirmedTxs gets number of unconfirmed transactions.

internal/rpc/core/routes.go

@@ -55,7 +55,7 @@ func NewRoutesMap(svc RPCService, opts *RouteOptions) RoutesMap {
 		"dump_consensus_state": rpc.NewRPCFunc(svc.DumpConsensusState),
 		"consensus_state":      rpc.NewRPCFunc(svc.GetConsensusState),
 		"consensus_params":     rpc.NewRPCFunc(svc.ConsensusParams, "height"),
-		"unconfirmed_txs":      rpc.NewRPCFunc(svc.UnconfirmedTxs, "limit"),
+		"unconfirmed_txs":      rpc.NewRPCFunc(svc.UnconfirmedTxs, "page", "per_page"),
 		"num_unconfirmed_txs":  rpc.NewRPCFunc(svc.NumUnconfirmedTxs),
 
 		// tx broadcast API
@@ -107,7 +107,7 @@ type RPCService interface {
 	Subscribe(ctx context.Context, query string) (*coretypes.ResultSubscribe, error)
 	Tx(ctx context.Context, hash bytes.HexBytes, prove bool) (*coretypes.ResultTx, error)
 	TxSearch(ctx context.Context, query string, prove bool, pagePtr, perPagePtr *int, orderBy string) (*coretypes.ResultTxSearch, error)
-	UnconfirmedTxs(ctx context.Context, limitPtr *int) (*coretypes.ResultUnconfirmedTxs, error)
+	UnconfirmedTxs(ctx context.Context, page, perPage *int) (*coretypes.ResultUnconfirmedTxs, error)
 	Unsubscribe(ctx context.Context, query string) (*coretypes.ResultUnsubscribe, error)
 	UnsubscribeAll(ctx context.Context) (*coretypes.ResultUnsubscribe, error)
 	Validators(ctx context.Context, heightPtr *int64, pagePtr, perPagePtr *int) (*coretypes.ResultValidators, error)

internal/statesync/dispatcher.go

@@ -195,7 +195,7 @@ func (p *BlockProvider) LightBlock(ctx context.Context, height int64) (*types.Li
 	case errPeerAlreadyBusy:
 		return nil, provider.ErrLightBlockNotFound
 	default:
-		return nil, provider.ErrUnreliableProvider{Reason: err.Error()}
+		return nil, provider.ErrUnreliableProvider{Reason: err}
 	}
 
 	// check that the height requested is the same one returned