rpc/client: take context as first param (#5347)

Closes #5145

also applies to light/client

statesync/mocks/state_provider.go

@@ -3,6 +3,8 @@
 package mocks
 
 import (
+	context "context"
+
 	mock "github.com/stretchr/testify/mock"
 	state "github.com/tendermint/tendermint/state"
 
@@ -14,13 +16,13 @@ type StateProvider struct {
 	mock.Mock
 }
 
-// AppHash provides a mock function with given fields: height
-func (_m *StateProvider) AppHash(height uint64) ([]byte, error) {
-	ret := _m.Called(height)
+// AppHash provides a mock function with given fields: ctx, height
+func (_m *StateProvider) AppHash(ctx context.Context, height uint64) ([]byte, error) {
+	ret := _m.Called(ctx, height)
 
 	var r0 []byte
-	if rf, ok := ret.Get(0).(func(uint64) []byte); ok {
-		r0 = rf(height)
+	if rf, ok := ret.Get(0).(func(context.Context, uint64) []byte); ok {
+		r0 = rf(ctx, height)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]byte)
@@ -28,8 +30,8 @@ func (_m *StateProvider) AppHash(height uint64) ([]byte, error) {
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(uint64) error); ok {
-		r1 = rf(height)
+	if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok {
+		r1 = rf(ctx, height)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -37,13 +39,13 @@ func (_m *StateProvider) AppHash(height uint64) ([]byte, error) {
 	return r0, r1
 }
 
-// Commit provides a mock function with given fields: height
-func (_m *StateProvider) Commit(height uint64) (*types.Commit, error) {
-	ret := _m.Called(height)
+// Commit provides a mock function with given fields: ctx, height
+func (_m *StateProvider) Commit(ctx context.Context, height uint64) (*types.Commit, error) {
+	ret := _m.Called(ctx, height)
 
 	var r0 *types.Commit
-	if rf, ok := ret.Get(0).(func(uint64) *types.Commit); ok {
-		r0 = rf(height)
+	if rf, ok := ret.Get(0).(func(context.Context, uint64) *types.Commit); ok {
+		r0 = rf(ctx, height)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(*types.Commit)
@@ -51,8 +53,8 @@ func (_m *StateProvider) Commit(height uint64) (*types.Commit, error) {
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(uint64) error); ok {
-		r1 = rf(height)
+	if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok {
+		r1 = rf(ctx, height)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -60,20 +62,20 @@ func (_m *StateProvider) Commit(height uint64) (*types.Commit, error) {
 	return r0, r1
 }
 
-// State provides a mock function with given fields: height
-func (_m *StateProvider) State(height uint64) (state.State, error) {
-	ret := _m.Called(height)
+// State provides a mock function with given fields: ctx, height
+func (_m *StateProvider) State(ctx context.Context, height uint64) (state.State, error) {
+	ret := _m.Called(ctx, height)
 
 	var r0 state.State
-	if rf, ok := ret.Get(0).(func(uint64) state.State); ok {
-		r0 = rf(height)
+	if rf, ok := ret.Get(0).(func(context.Context, uint64) state.State); ok {
+		r0 = rf(ctx, height)
 	} else {
 		r0 = ret.Get(0).(state.State)
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(uint64) error); ok {
-		r1 = rf(height)
+	if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok {
+		r1 = rf(ctx, height)
 	} else {
 		r1 = ret.Error(1)
 	}

statesync/snapshots.go

@@ -1,10 +1,12 @@
 package statesync
 
 import (
+	"context"
 	"crypto/sha256"
 	"fmt"
 	"math/rand"
 	"sort"
+	"time"
 
 	tmsync "github.com/tendermint/tendermint/libs/sync"
 	"github.com/tendermint/tendermint/p2p"
@@ -76,7 +78,10 @@ func newSnapshotPool(stateProvider StateProvider) *snapshotPool {
 // returns true if this was a new, non-blacklisted snapshot. The snapshot height is verified using
 // the light client, and the expected app hash is set for the snapshot.
 func (p *snapshotPool) Add(peer p2p.Peer, snapshot *snapshot) (bool, error) {
-	appHash, err := p.stateProvider.AppHash(snapshot.Height)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	appHash, err := p.stateProvider.AppHash(ctx, snapshot.Height)
 	if err != nil {
 		return false, err
 	}

statesync/snapshots_test.go

@@ -42,7 +42,7 @@ func TestSnapshot_Key(t *testing.T) {
 
 func TestSnapshotPool_Add(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", uint64(1)).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, uint64(1)).Return([]byte("app_hash"), nil)
 
 	peer := &p2pmocks.Peer{}
 	peer.On("ID").Return(p2p.ID("id"))
@@ -80,7 +80,7 @@ func TestSnapshotPool_Add(t *testing.T) {
 
 func TestSnapshotPool_GetPeer(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	pool := newSnapshotPool(stateProvider)
 
 	s := &snapshot{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1}}
@@ -116,7 +116,7 @@ func TestSnapshotPool_GetPeer(t *testing.T) {
 
 func TestSnapshotPool_GetPeers(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	pool := newSnapshotPool(stateProvider)
 
 	s := &snapshot{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1}}
@@ -140,7 +140,7 @@ func TestSnapshotPool_GetPeers(t *testing.T) {
 
 func TestSnapshotPool_Ranked_Best(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	pool := newSnapshotPool(stateProvider)
 
 	// snapshots in expected order (best to worst). Highest height wins, then highest format.
@@ -185,7 +185,7 @@ func TestSnapshotPool_Ranked_Best(t *testing.T) {
 
 func TestSnapshotPool_Reject(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	pool := newSnapshotPool(stateProvider)
 	peer := &p2pmocks.Peer{}
 	peer.On("ID").Return(p2p.ID("id"))
@@ -215,7 +215,7 @@ func TestSnapshotPool_Reject(t *testing.T) {
 
 func TestSnapshotPool_RejectFormat(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	pool := newSnapshotPool(stateProvider)
 	peer := &p2pmocks.Peer{}
 	peer.On("ID").Return(p2p.ID("id"))
@@ -246,7 +246,7 @@ func TestSnapshotPool_RejectFormat(t *testing.T) {
 
 func TestSnapshotPool_RejectPeer(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	pool := newSnapshotPool(stateProvider)
 
 	peerA := &p2pmocks.Peer{}
@@ -288,7 +288,7 @@ func TestSnapshotPool_RejectPeer(t *testing.T) {
 
 func TestSnapshotPool_RemovePeer(t *testing.T) {
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	pool := newSnapshotPool(stateProvider)
 
 	peerA := &p2pmocks.Peer{}

statesync/stateprovider.go

@@ -1,6 +1,7 @@
 package statesync
 
 import (
+	"context"
 	"fmt"
 	"strings"
 	"time"
@@ -26,11 +27,11 @@ import (
 // to the state.State object, not the state machine.
 type StateProvider interface {
 	// AppHash returns the app hash after the given height has been committed.
-	AppHash(height uint64) ([]byte, error)
+	AppHash(ctx context.Context, height uint64) ([]byte, error)
 	// Commit returns the commit at the given height.
-	Commit(height uint64) (*types.Commit, error)
+	Commit(ctx context.Context, height uint64) (*types.Commit, error)
 	// State returns a state object at the given height.
-	State(height uint64) (sm.State, error)
+	State(ctx context.Context, height uint64) (sm.State, error)
 }
 
 // lightClientStateProvider is a state provider using the light client.
@@ -44,6 +45,7 @@ type lightClientStateProvider struct {
 
 // NewLightClientStateProvider creates a new StateProvider using a light client and RPC clients.
 func NewLightClientStateProvider(
+	ctx context.Context,
 	chainID string,
 	version tmstate.Version,
 	initialHeight int64,
@@ -69,7 +71,7 @@ func NewLightClientStateProvider(
 		providerRemotes[provider] = server
 	}
 
-	lc, err := light.NewClient(chainID, trustOptions, providers[0], providers[1:],
+	lc, err := light.NewClient(ctx, chainID, trustOptions, providers[0], providers[1:],
 		lightdb.New(dbm.NewMemDB(), ""), light.Logger(logger), light.MaxRetryAttempts(5))
 	if err != nil {
 		return nil, err
@@ -83,12 +85,12 @@ func NewLightClientStateProvider(
 }
 
 // AppHash implements StateProvider.
-func (s *lightClientStateProvider) AppHash(height uint64) ([]byte, error) {
+func (s *lightClientStateProvider) AppHash(ctx context.Context, height uint64) ([]byte, error) {
 	s.Lock()
 	defer s.Unlock()
 
 	// We have to fetch the next height, which contains the app hash for the previous height.
-	header, err := s.lc.VerifyLightBlockAtHeight(int64(height+1), time.Now())
+	header, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height+1), time.Now())
 	if err != nil {
 		return nil, err
 	}
@@ -96,10 +98,10 @@ func (s *lightClientStateProvider) AppHash(height uint64) ([]byte, error) {
 }
 
 // Commit implements StateProvider.
-func (s *lightClientStateProvider) Commit(height uint64) (*types.Commit, error) {
+func (s *lightClientStateProvider) Commit(ctx context.Context, height uint64) (*types.Commit, error) {
 	s.Lock()
 	defer s.Unlock()
-	header, err := s.lc.VerifyLightBlockAtHeight(int64(height), time.Now())
+	header, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height), time.Now())
 	if err != nil {
 		return nil, err
 	}
@@ -107,7 +109,7 @@ func (s *lightClientStateProvider) Commit(height uint64) (*types.Commit, error)
 }
 
 // State implements StateProvider.
-func (s *lightClientStateProvider) State(height uint64) (sm.State, error) {
+func (s *lightClientStateProvider) State(ctx context.Context, height uint64) (sm.State, error) {
 	s.Lock()
 	defer s.Unlock()
 
@@ -128,15 +130,15 @@ func (s *lightClientStateProvider) State(height uint64) (sm.State, error) {
 	//
 	// We need to fetch the NextValidators from height+2 because if the application changed
 	// the validator set at the snapshot height then this only takes effect at height+2.
-	lastLightBlock, err := s.lc.VerifyLightBlockAtHeight(int64(height), time.Now())
+	lastLightBlock, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height), time.Now())
 	if err != nil {
 		return sm.State{}, err
 	}
-	curLightBlock, err := s.lc.VerifyLightBlockAtHeight(int64(height+1), time.Now())
+	curLightBlock, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height+1), time.Now())
 	if err != nil {
 		return sm.State{}, err
 	}
-	nextLightBlock, err := s.lc.VerifyLightBlockAtHeight(int64(height+2), time.Now())
+	nextLightBlock, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height+2), time.Now())
 	if err != nil {
 		return sm.State{}, err
 	}
@@ -161,7 +163,7 @@ func (s *lightClientStateProvider) State(height uint64) (sm.State, error) {
 		return sm.State{}, fmt.Errorf("unable to create RPC client: %w", err)
 	}
 	rpcclient := lightrpc.NewClient(primaryRPC, s.lc)
-	result, err := rpcclient.ConsensusParams(&nextLightBlock.Height)
+	result, err := rpcclient.ConsensusParams(ctx, &nextLightBlock.Height)
 	if err != nil {
 		return sm.State{}, fmt.Errorf("unable to fetch consensus parameters for height %v: %w",
 			nextLightBlock.Height, err)

statesync/syncer.go

@@ -241,12 +241,15 @@ func (s *syncer) Sync(snapshot *snapshot, chunks *chunkQueue) (sm.State, *types.
 		go s.fetchChunks(ctx, snapshot, chunks)
 	}
 
+	pctx, pcancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer pcancel()
+
 	// Optimistically build new state, so we don't discover any light client failures at the end.
-	state, err := s.stateProvider.State(snapshot.Height)
+	state, err := s.stateProvider.State(pctx, snapshot.Height)
 	if err != nil {
 		return sm.State{}, nil, fmt.Errorf("failed to build new state: %w", err)
 	}
-	commit, err := s.stateProvider.Commit(snapshot.Height)
+	commit, err := s.stateProvider.Commit(pctx, snapshot.Height)
 	if err != nil {
 		return sm.State{}, nil, fmt.Errorf("failed to fetch commit: %w", err)
 	}

statesync/syncer_test.go

@@ -30,7 +30,7 @@ func setupOfferSyncer(t *testing.T) (*syncer, *proxymocks.AppConnSnapshot) {
 	connQuery := &proxymocks.AppConnQuery{}
 	connSnapshot := &proxymocks.AppConnSnapshot{}
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 	syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
 	return syncer, connSnapshot
 }
@@ -77,10 +77,10 @@ func TestSyncer_SyncAny(t *testing.T) {
 	s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
 
 	stateProvider := &mocks.StateProvider{}
-	stateProvider.On("AppHash", uint64(1)).Return(state.AppHash, nil)
-	stateProvider.On("AppHash", uint64(2)).Return([]byte("app_hash_2"), nil)
-	stateProvider.On("Commit", uint64(1)).Return(commit, nil)
-	stateProvider.On("State", uint64(1)).Return(state, nil)
+	stateProvider.On("AppHash", mock.Anything, uint64(1)).Return(state.AppHash, nil)
+	stateProvider.On("AppHash", mock.Anything, uint64(2)).Return([]byte("app_hash_2"), nil)
+	stateProvider.On("Commit", mock.Anything, uint64(1)).Return(commit, nil)
+	stateProvider.On("State", mock.Anything, uint64(1)).Return(state, nil)
 	connSnapshot := &proxymocks.AppConnSnapshot{}
 	connQuery := &proxymocks.AppConnQuery{}
 
@@ -406,7 +406,7 @@ func TestSyncer_applyChunks_Results(t *testing.T) {
 			connQuery := &proxymocks.AppConnQuery{}
 			connSnapshot := &proxymocks.AppConnSnapshot{}
 			stateProvider := &mocks.StateProvider{}
-			stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+			stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 			syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
 
 			body := []byte{1, 2, 3}
@@ -457,7 +457,7 @@ func TestSyncer_applyChunks_RefetchChunks(t *testing.T) {
 			connQuery := &proxymocks.AppConnQuery{}
 			connSnapshot := &proxymocks.AppConnSnapshot{}
 			stateProvider := &mocks.StateProvider{}
-			stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+			stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 			syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
 
 			chunks, err := newChunkQueue(&snapshot{Height: 1, Format: 1, Chunks: 3}, "")
@@ -520,7 +520,7 @@ func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
 			connQuery := &proxymocks.AppConnQuery{}
 			connSnapshot := &proxymocks.AppConnSnapshot{}
 			stateProvider := &mocks.StateProvider{}
-			stateProvider.On("AppHash", mock.Anything).Return([]byte("app_hash"), nil)
+			stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
 			syncer := newSyncer(log.NewNopLogger(), connSnapshot, connQuery, stateProvider, "")
 
 			// Set up three peers across two snapshots, and ask for one of them to be banned.