mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-27 10:24:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

make the rest of the code compile

Browse Source
This commit is contained in:
Callum Waters
2021-08-24 13:26:01 +02:00
parent dcf91478f8
commit 68c54f0676
283 changed files with 3874 additions and 3906 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/statesync/block_queue.go
+5 -4
View File
@@ -6,12 +6,13 @@ import (
"sync"
"time"
"github.com/tendermint/tendermint/types"
"github.com/tendermint/tendermint/pkg/light"
"github.com/tendermint/tendermint/pkg/p2p"
)
type lightBlockResponse struct {
block *types.LightBlock
peer types.NodeID
block *light.LightBlock
peer p2p.NodeID
}
// a block queue is used for asynchronously fetching and verifying light blocks
@@ -26,7 +27,7 @@ type blockQueue struct {
initialHeight int64
stopHeight int64
stopTime time.Time
terminal *types.LightBlock
terminal *light.LightBlock
// track failed heights so we know what blocks to try fetch again
failed *maxIntHeap
internal/statesync/block_queue_test.go
+8 -8
View File
@@ -10,7 +10,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/internal/test/factory"
"github.com/tendermint/tendermint/types"
"github.com/tendermint/tendermint/pkg/p2p"
)
var (
@@ -22,7 +22,7 @@ var (
)
func TestBlockQueueBasic(t *testing.T) {
peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
queue := newBlockQueue(startHeight, stopHeight, 1, stopTime, 1)
@@ -69,7 +69,7 @@ loop:
// Test with spurious failures and retries
func TestBlockQueueWithFailures(t *testing.T) {
peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
queue := newBlockQueue(startHeight, stopHeight, 1, stopTime, 200)
@@ -119,7 +119,7 @@ func TestBlockQueueWithFailures(t *testing.T) {
// Test that when all the blocks are retrieved that the queue still holds on to
// it's workers and in the event of failure can still fetch the failed block
func TestBlockQueueBlocks(t *testing.T) {
peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
queue := newBlockQueue(startHeight, stopHeight, 1, stopTime, 2)
expectedHeight := startHeight
@@ -166,7 +166,7 @@ loop:
}
func TestBlockQueueAcceptsNoMoreBlocks(t *testing.T) {
peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
queue := newBlockQueue(startHeight, stopHeight, 1, stopTime, 1)
defer queue.close()
@@ -191,7 +191,7 @@ loop:
// Test a scenario where more blocks are needed then just the stopheight because
// we haven't found a block with a small enough time.
func TestBlockQueueStopTime(t *testing.T) {
peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
queue := newBlockQueue(startHeight, stopHeight, 1, stopTime, 1)
@@ -234,7 +234,7 @@ func TestBlockQueueStopTime(t *testing.T) {
}
func TestBlockQueueInitialHeight(t *testing.T) {
peerID, err := types.NewNodeID("0011223344556677889900112233445566778899")
peerID, err := p2p.NewNodeID("0011223344556677889900112233445566778899")
require.NoError(t, err)
const initialHeight int64 = 120
@@ -273,7 +273,7 @@ loop:
}
}
func mockLBResp(t *testing.T, peer types.NodeID, height int64, time time.Time) lightBlockResponse {
func mockLBResp(t *testing.T, peer p2p.NodeID, height int64, time time.Time) lightBlockResponse {
return lightBlockResponse{
block: mockLB(t, height, time, factory.MakeBlockID()),
peer: peer,
internal/statesync/chunks.go
+6 -6
View File
@@ -10,7 +10,7 @@ import (
"time"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/types"
"github.com/tendermint/tendermint/pkg/p2p"
)
// errDone is returned by chunkQueue.Next() when all chunks have been returned.
@@ -22,7 +22,7 @@ type chunk struct {
Format uint32
Index uint32
Chunk []byte
Sender types.NodeID
Sender p2p.NodeID
}
// chunkQueue manages chunks for a state sync process, ordering them if requested. It acts as an
@@ -33,7 +33,7 @@ type chunkQueue struct {
snapshot *snapshot // if this is nil, the queue has been closed
dir string // temp dir for on-disk chunk storage
chunkFiles map[uint32]string // path to temporary chunk file
chunkSenders map[uint32]types.NodeID // the peer who sent the given chunk
chunkSenders map[uint32]p2p.NodeID // the peer who sent the given chunk
chunkAllocated map[uint32]bool // chunks that have been allocated via Allocate()
chunkReturned map[uint32]bool // chunks returned via Next()
waiters map[uint32][]chan<- uint32 // signals WaitFor() waiters about chunk arrival
@@ -54,7 +54,7 @@ func newChunkQueue(snapshot *snapshot, tempDir string) (*chunkQueue, error) {
snapshot: snapshot,
dir: dir,
chunkFiles: make(map[uint32]string, snapshot.Chunks),
chunkSenders: make(map[uint32]types.NodeID, snapshot.Chunks),
chunkSenders: make(map[uint32]p2p.NodeID, snapshot.Chunks),
chunkAllocated: make(map[uint32]bool, snapshot.Chunks),
chunkReturned: make(map[uint32]bool, snapshot.Chunks),
waiters: make(map[uint32][]chan<- uint32),
@@ -188,7 +188,7 @@ func (q *chunkQueue) discard(index uint32) error {
// DiscardSender discards all *unreturned* chunks from a given sender. If the caller wants to
// discard already returned chunks, this can be done via Discard().
func (q *chunkQueue) DiscardSender(peerID types.NodeID) error {
func (q *chunkQueue) DiscardSender(peerID p2p.NodeID) error {
q.Lock()
defer q.Unlock()
@@ -208,7 +208,7 @@ func (q *chunkQueue) DiscardSender(peerID types.NodeID) error {
// GetSender returns the sender of the chunk with the given index, or empty if
// not found.
func (q *chunkQueue) GetSender(index uint32) types.NodeID {
func (q *chunkQueue) GetSender(index uint32) p2p.NodeID {
q.Lock()
defer q.Unlock()
return q.chunkSenders[index]
internal/statesync/chunks_test.go
+16 -16
View File
@@ -8,7 +8,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/types"
"github.com/tendermint/tendermint/pkg/p2p"
)
func setupChunkQueue(t *testing.T) (*chunkQueue, func()) {
@@ -274,7 +274,7 @@ func TestChunkQueue_DiscardSender(t *testing.T) {
defer teardown()
// Allocate and add all chunks to the queue
senders := []types.NodeID{types.NodeID("a"), types.NodeID("b"), types.NodeID("c")}
senders := []p2p.NodeID{p2p.NodeID("a"), p2p.NodeID("b"), p2p.NodeID("c")}
for i := uint32(0); i < queue.Size(); i++ {
_, err := queue.Allocate()
require.NoError(t, err)
@@ -295,14 +295,14 @@ func TestChunkQueue_DiscardSender(t *testing.T) {
}
// Discarding an unknown sender should do nothing
err := queue.DiscardSender(types.NodeID("x"))
err := queue.DiscardSender(p2p.NodeID("x"))
require.NoError(t, err)
_, err = queue.Allocate()
assert.Equal(t, errDone, err)
// Discarding sender b should discard chunk 4, but not chunk 1 which has already been
// returned.
err = queue.DiscardSender(types.NodeID("b"))
err = queue.DiscardSender(p2p.NodeID("b"))
require.NoError(t, err)
index, err := queue.Allocate()
require.NoError(t, err)
@@ -315,8 +315,8 @@ func TestChunkQueue_GetSender(t *testing.T) {
queue, teardown := setupChunkQueue(t)
defer teardown()
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
_, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{1}, Sender: peerAID})
require.NoError(t, err)
@@ -354,7 +354,7 @@ func TestChunkQueue_Next(t *testing.T) {
}()
assert.Empty(t, chNext)
_, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{3, 1, 1}, Sender: types.NodeID("b")})
_, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{3, 1, 1}, Sender: p2p.NodeID("b")})
require.NoError(t, err)
select {
case <-chNext:
@@ -362,17 +362,17 @@ func TestChunkQueue_Next(t *testing.T) {
default:
}
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}, Sender: types.NodeID("a")})
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}, Sender: p2p.NodeID("a")})
require.NoError(t, err)
assert.Equal(t,
&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}, Sender: types.NodeID("a")},
&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}, Sender: p2p.NodeID("a")},
<-chNext)
assert.Equal(t,
&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{3, 1, 1}, Sender: types.NodeID("b")},
&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{3, 1, 1}, Sender: p2p.NodeID("b")},
<-chNext)
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 4, Chunk: []byte{3, 1, 4}, Sender: types.NodeID("e")})
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 4, Chunk: []byte{3, 1, 4}, Sender: p2p.NodeID("e")})
require.NoError(t, err)
select {
case <-chNext:
@@ -380,19 +380,19 @@ func TestChunkQueue_Next(t *testing.T) {
default:
}
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 2, Chunk: []byte{3, 1, 2}, Sender: types.NodeID("c")})
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 2, Chunk: []byte{3, 1, 2}, Sender: p2p.NodeID("c")})
require.NoError(t, err)
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 3, Chunk: []byte{3, 1, 3}, Sender: types.NodeID("d")})
_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 3, Chunk: []byte{3, 1, 3}, Sender: p2p.NodeID("d")})
require.NoError(t, err)
assert.Equal(t,
&chunk{Height: 3, Format: 1, Index: 2, Chunk: []byte{3, 1, 2}, Sender: types.NodeID("c")},
&chunk{Height: 3, Format: 1, Index: 2, Chunk: []byte{3, 1, 2}, Sender: p2p.NodeID("c")},
<-chNext)
assert.Equal(t,
&chunk{Height: 3, Format: 1, Index: 3, Chunk: []byte{3, 1, 3}, Sender: types.NodeID("d")},
&chunk{Height: 3, Format: 1, Index: 3, Chunk: []byte{3, 1, 3}, Sender: p2p.NodeID("d")},
<-chNext)
assert.Equal(t,
&chunk{Height: 3, Format: 1, Index: 4, Chunk: []byte{3, 1, 4}, Sender: types.NodeID("e")},
&chunk{Height: 3, Format: 1, Index: 4, Chunk: []byte{3, 1, 4}, Sender: p2p.NodeID("e")},
<-chNext)
_, ok := <-chNext
internal/statesync/dispatcher.go
+26 -24
View File
@@ -9,9 +9,11 @@ import (
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/light/provider"
"github.com/tendermint/tendermint/pkg/evidence"
"github.com/tendermint/tendermint/pkg/light"
p2ptypes "github.com/tendermint/tendermint/pkg/p2p"
ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
proto "github.com/tendermint/tendermint/proto/tendermint/types"
"github.com/tendermint/tendermint/types"
)
var (
@@ -31,7 +33,7 @@ type dispatcher struct {
timeout time.Duration
mtx sync.Mutex
calls map[types.NodeID]chan *types.LightBlock
calls map[p2ptypes.NodeID]chan *light.LightBlock
running bool
}
@@ -40,14 +42,14 @@ func newDispatcher(requestCh chan<- p2p.Envelope, timeout time.Duration) *dispat
availablePeers: newPeerList(),
timeout: timeout,
requestCh: requestCh,
calls: make(map[types.NodeID]chan *types.LightBlock),
calls: make(map[p2ptypes.NodeID]chan *light.LightBlock),
running: true,
}
}
// LightBlock uses the request channel to fetch a light block from the next peer
// in a list, tracks the call and waits for the reactor to pass along the response
func (d *dispatcher) LightBlock(ctx context.Context, height int64) (*types.LightBlock, types.NodeID, error) {
func (d *dispatcher) LightBlock(ctx context.Context, height int64) (*light.LightBlock, p2ptypes.NodeID, error) {
d.mtx.Lock()
// check to see that the dispatcher is connected to at least one peer
if d.availablePeers.Len() == 0 && len(d.calls) == 0 {
@@ -98,7 +100,7 @@ func (d *dispatcher) start() {
d.running = true
}
func (d *dispatcher) lightBlock(ctx context.Context, height int64, peer types.NodeID) (*types.LightBlock, error) {
func (d *dispatcher) lightBlock(ctx context.Context, height int64, peer p2ptypes.NodeID) (*light.LightBlock, error) {
// dispatch the request to the peer
callCh, err := d.dispatch(peer, height)
if err != nil {
@@ -122,7 +124,7 @@ func (d *dispatcher) lightBlock(ctx context.Context, height int64, peer types.No
// respond allows the underlying process which receives requests on the
// requestCh to respond with the respective light block
func (d *dispatcher) respond(lb *proto.LightBlock, peer types.NodeID) error {
func (d *dispatcher) respond(lb *proto.LightBlock, peer p2ptypes.NodeID) error {
d.mtx.Lock()
defer d.mtx.Unlock()
@@ -142,7 +144,7 @@ func (d *dispatcher) respond(lb *proto.LightBlock, peer types.NodeID) error {
return nil
}
block, err := types.LightBlockFromProto(lb)
block, err := light.LightBlockFromProto(lb)
if err != nil {
fmt.Println("error with converting light block")
return err
@@ -152,11 +154,11 @@ func (d *dispatcher) respond(lb *proto.LightBlock, peer types.NodeID) error {
return nil
}
func (d *dispatcher) addPeer(peer types.NodeID) {
func (d *dispatcher) addPeer(peer p2ptypes.NodeID) {
d.availablePeers.Append(peer)
}
func (d *dispatcher) removePeer(peer types.NodeID) {
func (d *dispatcher) removePeer(peer p2ptypes.NodeID) {
d.mtx.Lock()
defer d.mtx.Unlock()
if _, ok := d.calls[peer]; ok {
@@ -168,10 +170,10 @@ func (d *dispatcher) removePeer(peer types.NodeID) {
// dispatch takes a peer and allocates it a channel so long as it's not already
// busy and the receiving channel is still running. It then dispatches the message
func (d *dispatcher) dispatch(peer types.NodeID, height int64) (chan *types.LightBlock, error) {
func (d *dispatcher) dispatch(peer p2ptypes.NodeID, height int64) (chan *light.LightBlock, error) {
d.mtx.Lock()
defer d.mtx.Unlock()
ch := make(chan *types.LightBlock, 1)
ch := make(chan *light.LightBlock, 1)
// check if the dispatcher is running or not
if !d.running {
@@ -198,7 +200,7 @@ func (d *dispatcher) dispatch(peer types.NodeID, height int64) (chan *types.Ligh
// release appends the peer back to the list and deletes the allocated call so
// that a new call can be made to that peer
func (d *dispatcher) release(peer types.NodeID) {
func (d *dispatcher) release(peer p2ptypes.NodeID) {
d.mtx.Lock()
defer d.mtx.Unlock()
if call, ok := d.calls[peer]; ok {
@@ -216,13 +218,13 @@ func (d *dispatcher) release(peer types.NodeID) {
// TODO: This should probably be moved over to the light package but as we're
// not yet officially supporting p2p light clients we'll leave this here for now.
type blockProvider struct {
peer types.NodeID
peer p2ptypes.NodeID
chainID string
timeout time.Duration
dispatcher *dispatcher
}
func (p *blockProvider) LightBlock(ctx context.Context, height int64) (*types.LightBlock, error) {
func (p *blockProvider) LightBlock(ctx context.Context, height int64) (*light.LightBlock, error) {
// FIXME: The provider doesn't know if the dispatcher is still connected to
// that peer. If the connection is dropped for whatever reason the
// dispatcher needs to be able to relay this back to the provider so it can
@@ -245,7 +247,7 @@ func (p *blockProvider) LightBlock(ctx context.Context, height int64) (*types.Li
// attacks. This is a no op as there currently isn't a way to wire this up to
// the evidence reactor (we should endeavor to do this in the future but for now
// it's not critical for backwards verification)
func (p *blockProvider) ReportEvidence(ctx context.Context, ev types.Evidence) error {
func (p *blockProvider) ReportEvidence(ctx context.Context, ev evidence.Evidence) error {
return nil
}
@@ -258,14 +260,14 @@ func (p *blockProvider) String() string { return string(p.peer) }
// retrieving blocks over all the peers the reactor is connected to
type peerlist struct {
mtx sync.Mutex
peers []types.NodeID
waiting []chan types.NodeID
peers []p2ptypes.NodeID
waiting []chan p2ptypes.NodeID
}
func newPeerList() *peerlist {
return &peerlist{
peers: make([]types.NodeID, 0),
waiting: make([]chan types.NodeID, 0),
peers: make([]p2ptypes.NodeID, 0),
waiting: make([]chan p2ptypes.NodeID, 0),
}
}
@@ -275,12 +277,12 @@ func (l *peerlist) Len() int {
return len(l.peers)
}
func (l *peerlist) Pop(ctx context.Context) types.NodeID {
func (l *peerlist) Pop(ctx context.Context) p2ptypes.NodeID {
l.mtx.Lock()
if len(l.peers) == 0 {
// if we don't have any peers in the list we block until a peer is
// appended
wait := make(chan types.NodeID, 1)
wait := make(chan p2ptypes.NodeID, 1)
l.waiting = append(l.waiting, wait)
// unlock whilst waiting so that the list can be appended to
l.mtx.Unlock()
@@ -299,7 +301,7 @@ func (l *peerlist) Pop(ctx context.Context) types.NodeID {
return peer
}
func (l *peerlist) Append(peer types.NodeID) {
func (l *peerlist) Append(peer p2ptypes.NodeID) {
l.mtx.Lock()
defer l.mtx.Unlock()
if len(l.waiting) > 0 {
@@ -312,7 +314,7 @@ func (l *peerlist) Append(peer types.NodeID) {
}
}
func (l *peerlist) Remove(peer types.NodeID) {
func (l *peerlist) Remove(peer p2ptypes.NodeID) {
l.mtx.Lock()
defer l.mtx.Unlock()
for i, p := range l.peers {
@@ -323,7 +325,7 @@ func (l *peerlist) Remove(peer types.NodeID) {
}
}
func (l *peerlist) Peers() []types.NodeID {
func (l *peerlist) Peers() []p2ptypes.NodeID {
l.mtx.Lock()
defer l.mtx.Unlock()
return l.peers
internal/statesync/dispatcher_test.go
+7 -6
View File
@@ -13,8 +13,9 @@ import (
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/pkg/light"
p2ptypes "github.com/tendermint/tendermint/pkg/p2p"
ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
"github.com/tendermint/tendermint/types"
)
func TestDispatcherBasic(t *testing.T) {
@@ -112,7 +113,7 @@ func TestDispatcherReturnsBlockOncePeerAvailable(t *testing.T) {
go func() {
<-dispatcherRequestCh
lb := &types.LightBlock{}
lb := &light.LightBlock{}
asProto, err := lb.ToProto()
require.Nil(t, err)
err = d.respond(asProto, peerFromSet)
@@ -178,7 +179,7 @@ func TestPeerListBasic(t *testing.T) {
}
assert.Equal(t, half, peerList.Len())
peerList.Remove(types.NodeID("lp"))
peerList.Remove(p2ptypes.NodeID("lp"))
assert.Equal(t, half, peerList.Len())
peerList.Remove(peerSet[half])
@@ -295,10 +296,10 @@ func handleRequests(t *testing.T, d *dispatcher, ch chan p2p.Envelope, closeCh c
}
}
func createPeerSet(num int) []types.NodeID {
peers := make([]types.NodeID, num)
func createPeerSet(num int) []p2ptypes.NodeID {
peers := make([]p2ptypes.NodeID, num)
for i := 0; i < num; i++ {
peers[i], _ = types.NewNodeID(strings.Repeat(fmt.Sprintf("%d", i), 2*types.NodeIDByteLength))
peers[i], _ = p2ptypes.NewNodeID(strings.Repeat(fmt.Sprintf("%d", i), 2*p2ptypes.NodeIDByteLength))
}
return peers
}
internal/statesync/mocks/state_provider.go
+6 -6
View File
@@ -6,9 +6,9 @@ import (
context "context"
mock "github.com/stretchr/testify/mock"
state "github.com/tendermint/tendermint/state"
metadata "github.com/tendermint/tendermint/pkg/metadata"
types "github.com/tendermint/tendermint/types"
state "github.com/tendermint/tendermint/state"
)
// StateProvider is an autogenerated mock type for the StateProvider type
@@ -40,15 +40,15 @@ func (_m *StateProvider) AppHash(ctx context.Context, height uint64) ([]byte, er
}
// Commit provides a mock function with given fields: ctx, height
func (_m *StateProvider) Commit(ctx context.Context, height uint64) (*types.Commit, error) {
func (_m *StateProvider) Commit(ctx context.Context, height uint64) (*metadata.Commit, error) {
ret := _m.Called(ctx, height)
var r0 *types.Commit
if rf, ok := ret.Get(0).(func(context.Context, uint64) *types.Commit); ok {
var r0 *metadata.Commit
if rf, ok := ret.Get(0).(func(context.Context, uint64) *metadata.Commit); ok {
r0 = rf(ctx, height)
} else {
if ret.Get(0) != nil {
r0 = ret.Get(0).(*types.Commit)
r0 = ret.Get(0).(*metadata.Commit)
}
}
internal/statesync/reactor.go
+9 -7
View File
@@ -10,17 +10,19 @@ import (
"sort"
"time"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/config"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/libs/service"
"github.com/tendermint/tendermint/pkg/abci"
"github.com/tendermint/tendermint/pkg/consensus"
"github.com/tendermint/tendermint/pkg/light"
"github.com/tendermint/tendermint/pkg/metadata"
ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
"github.com/tendermint/tendermint/proxy"
sm "github.com/tendermint/tendermint/state"
"github.com/tendermint/tendermint/store"
"github.com/tendermint/tendermint/types"
)
var (
@@ -304,7 +306,7 @@ func (r *Reactor) backfill(
ctx context.Context,
chainID string,
startHeight, stopHeight, initialHeight int64,
trustedBlockID types.BlockID,
trustedBlockID metadata.BlockID,
stopTime time.Time,
) error {
r.Logger.Info("starting backfill process...", "startHeight", startHeight,
@@ -312,7 +314,7 @@ func (r *Reactor) backfill(
const sleepTime = 1 * time.Second
var (
lastValidatorSet *types.ValidatorSet
lastValidatorSet *consensus.ValidatorSet
lastChangeHeight int64 = startHeight
)
@@ -811,7 +813,7 @@ func (r *Reactor) recentSnapshots(n uint32) ([]*snapshot, error) {
// fetchLightBlock works out whether the node has a light block at a particular
// height and if so returns it so it can be gossiped to peers
func (r *Reactor) fetchLightBlock(height uint64) (*types.LightBlock, error) {
func (r *Reactor) fetchLightBlock(height uint64) (*light.LightBlock, error) {
h := int64(height)
blockMeta := r.blockStore.LoadBlockMeta(h)
@@ -832,8 +834,8 @@ func (r *Reactor) fetchLightBlock(height uint64) (*types.LightBlock, error) {
return nil, nil
}
return &types.LightBlock{
SignedHeader: &types.SignedHeader{
return &light.LightBlock{
SignedHeader: &metadata.SignedHeader{
Header: &blockMeta.Header,
Commit: commit,
},
internal/statesync/reactor_test.go
+30 -27
View File
@@ -12,19 +12,22 @@ import (
"github.com/stretchr/testify/require"
dbm "github.com/tendermint/tm-db"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/internal/statesync/mocks"
"github.com/tendermint/tendermint/internal/test/factory"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/light/provider"
"github.com/tendermint/tendermint/pkg/abci"
"github.com/tendermint/tendermint/pkg/consensus"
"github.com/tendermint/tendermint/pkg/light"
"github.com/tendermint/tendermint/pkg/metadata"
p2ptypes "github.com/tendermint/tendermint/pkg/p2p"
ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
tmproto "github.com/tendermint/tendermint/proto/tendermint/types"
proxymocks "github.com/tendermint/tendermint/proxy/mocks"
smmocks "github.com/tendermint/tendermint/state/mocks"
"github.com/tendermint/tendermint/store"
"github.com/tendermint/tendermint/types"
)
type reactorTestSuite struct {
@@ -166,7 +169,7 @@ func TestReactor_ChunkRequest_InvalidRequest(t *testing.T) {
rts := setup(t, nil, nil, nil, 2)
rts.chunkInCh <- p2p.Envelope{
From: types.NodeID("aa"),
From: p2ptypes.NodeID("aa"),
Message: &ssproto.SnapshotsRequest{},
}
@@ -174,7 +177,7 @@ func TestReactor_ChunkRequest_InvalidRequest(t *testing.T) {
require.Error(t, response.Err)
require.Empty(t, rts.chunkOutCh)
require.Contains(t, response.Err.Error(), "received unknown message")
require.Equal(t, types.NodeID("aa"), response.NodeID)
require.Equal(t, p2ptypes.NodeID("aa"), response.NodeID)
}
func TestReactor_ChunkRequest(t *testing.T) {
@@ -220,7 +223,7 @@ func TestReactor_ChunkRequest(t *testing.T) {
rts := setup(t, conn, nil, nil, 2)
rts.chunkInCh <- p2p.Envelope{
From: types.NodeID("aa"),
From: p2ptypes.NodeID("aa"),
Message: tc.request,
}
@@ -237,7 +240,7 @@ func TestReactor_SnapshotsRequest_InvalidRequest(t *testing.T) {
rts := setup(t, nil, nil, nil, 2)
rts.snapshotInCh <- p2p.Envelope{
From: types.NodeID("aa"),
From: p2ptypes.NodeID("aa"),
Message: &ssproto.ChunkRequest{},
}
@@ -245,7 +248,7 @@ func TestReactor_SnapshotsRequest_InvalidRequest(t *testing.T) {
require.Error(t, response.Err)
require.Empty(t, rts.snapshotOutCh)
require.Contains(t, response.Err.Error(), "received unknown message")
require.Equal(t, types.NodeID("aa"), response.NodeID)
require.Equal(t, p2ptypes.NodeID("aa"), response.NodeID)
}
func TestReactor_SnapshotsRequest(t *testing.T) {
@@ -297,7 +300,7 @@ func TestReactor_SnapshotsRequest(t *testing.T) {
rts := setup(t, conn, nil, nil, 100)
rts.snapshotInCh <- p2p.Envelope{
From: types.NodeID("aa"),
From: p2ptypes.NodeID("aa"),
Message: &ssproto.SnapshotsRequest{},
}
@@ -329,18 +332,18 @@ func TestReactor_LightBlockResponse(t *testing.T) {
blockID, factory.DefaultTestTime)
require.NoError(t, err)
sh := &types.SignedHeader{
sh := &metadata.SignedHeader{
Header: h,
Commit: &types.Commit{
Commit: &metadata.Commit{
Height: h.Height,
BlockID: blockID,
Signatures: []types.CommitSig{
Signatures: []metadata.CommitSig{
vote.CommitSig(),
},
},
}
lb := &types.LightBlock{
lb := &light.LightBlock{
SignedHeader: sh,
ValidatorSet: vals,
}
@@ -350,7 +353,7 @@ func TestReactor_LightBlockResponse(t *testing.T) {
rts.stateStore.On("LoadValidators", height).Return(vals, nil)
rts.blockInCh <- p2p.Envelope{
From: types.NodeID("aa"),
From: p2ptypes.NodeID("aa"),
Message: &ssproto.LightBlockRequest{
Height: 10,
},
@@ -359,10 +362,10 @@ func TestReactor_LightBlockResponse(t *testing.T) {
select {
case response := <-rts.blockOutCh:
require.Equal(t, types.NodeID("aa"), response.To)
require.Equal(t, p2ptypes.NodeID("aa"), response.To)
res, ok := response.Message.(*ssproto.LightBlockResponse)
require.True(t, ok)
receivedLB, err := types.LightBlockFromProto(res.LightBlock)
receivedLB, err := light.LightBlockFromProto(res.LightBlock)
require.NoError(t, err)
require.Equal(t, lb, receivedLB)
case <-time.After(1 * time.Second):
@@ -373,11 +376,11 @@ func TestReactor_LightBlockResponse(t *testing.T) {
func TestReactor_Dispatcher(t *testing.T) {
rts := setup(t, nil, nil, nil, 2)
rts.peerUpdateCh <- p2p.PeerUpdate{
NodeID: types.NodeID("aa"),
NodeID: p2ptypes.NodeID("aa"),
Status: p2p.PeerStatusUp,
}
rts.peerUpdateCh <- p2p.PeerUpdate{
NodeID: types.NodeID("bb"),
NodeID: p2ptypes.NodeID("bb"),
Status: p2p.PeerStatusUp,
}
@@ -436,14 +439,14 @@ func TestReactor_Backfill(t *testing.T) {
peers := []string{"a", "b", "c", "d"}
for _, peer := range peers {
rts.peerUpdateCh <- p2p.PeerUpdate{
NodeID: types.NodeID(peer),
NodeID: p2ptypes.NodeID(peer),
Status: p2p.PeerStatusUp,
}
}
trackingHeight := startHeight
rts.stateStore.On("SaveValidatorSets", mock.AnythingOfType("int64"), mock.AnythingOfType("int64"),
mock.AnythingOfType("*types.ValidatorSet")).Return(func(lh, uh int64, vals *types.ValidatorSet) error {
mock.AnythingOfType("*types.ValidatorSet")).Return(func(lh, uh int64, vals *consensus.ValidatorSet) error {
require.Equal(t, trackingHeight, lh)
require.Equal(t, lh, uh)
require.GreaterOrEqual(t, lh, stopHeight)
@@ -500,7 +503,7 @@ func retryUntil(t *testing.T, fn func() bool, timeout time.Duration) {
}
func handleLightBlockRequests(t *testing.T,
chain map[int64]*types.LightBlock,
chain map[int64]*light.LightBlock,
receiving chan p2p.Envelope,
sending chan p2p.Envelope,
close chan struct{},
@@ -550,8 +553,8 @@ func handleLightBlockRequests(t *testing.T,
}
}
func buildLightBlockChain(t *testing.T, fromHeight, toHeight int64, startTime time.Time) map[int64]*types.LightBlock {
chain := make(map[int64]*types.LightBlock, toHeight-fromHeight)
func buildLightBlockChain(t *testing.T, fromHeight, toHeight int64, startTime time.Time) map[int64]*light.LightBlock {
chain := make(map[int64]*light.LightBlock, toHeight-fromHeight)
lastBlockID := factory.MakeBlockID()
blockTime := startTime.Add(-5 * time.Minute)
for height := fromHeight; height < toHeight; height++ {
@@ -563,8 +566,8 @@ func buildLightBlockChain(t *testing.T, fromHeight, toHeight int64, startTime ti
}
func mockLB(t *testing.T, height int64, time time.Time,
lastBlockID types.BlockID) *types.LightBlock {
header, err := factory.MakeHeader(&types.Header{
lastBlockID metadata.BlockID) *light.LightBlock {
header, err := factory.MakeHeader(&metadata.Header{
Height: height,
LastBlockID: lastBlockID,
Time: time,
@@ -573,11 +576,11 @@ func mockLB(t *testing.T, height int64, time time.Time,
vals, pv := factory.RandValidatorSet(3, 10)
header.ValidatorsHash = vals.Hash()
lastBlockID = factory.MakeBlockIDWithHash(header.Hash())
voteSet := types.NewVoteSet(factory.DefaultTestChainID, height, 0, tmproto.PrecommitType, vals)
voteSet := consensus.NewVoteSet(factory.DefaultTestChainID, height, 0, tmproto.PrecommitType, vals)
commit, err := factory.MakeCommit(lastBlockID, height, 0, voteSet, pv, time)
require.NoError(t, err)
return &types.LightBlock{
SignedHeader: &types.SignedHeader{
return &light.LightBlock{
SignedHeader: &metadata.SignedHeader{
Header: header,
Commit: commit,
},
internal/statesync/snapshots.go
+15 -15
View File
@@ -10,7 +10,7 @@ import (
"time"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/types"
"github.com/tendermint/tendermint/pkg/p2p"
)
// snapshotKey is a snapshot key used for lookups.
@@ -47,16 +47,16 @@ type snapshotPool struct {
tmsync.Mutex
snapshots map[snapshotKey]*snapshot
snapshotPeers map[snapshotKey]map[types.NodeID]types.NodeID
snapshotPeers map[snapshotKey]map[p2p.NodeID]p2p.NodeID
// indexes for fast searches
formatIndex map[uint32]map[snapshotKey]bool
heightIndex map[uint64]map[snapshotKey]bool
peerIndex map[types.NodeID]map[snapshotKey]bool
peerIndex map[p2p.NodeID]map[snapshotKey]bool
// blacklists for rejected items
formatBlacklist map[uint32]bool
peerBlacklist map[types.NodeID]bool
peerBlacklist map[p2p.NodeID]bool
snapshotBlacklist map[snapshotKey]bool
}
@@ -65,12 +65,12 @@ func newSnapshotPool(stateProvider StateProvider) *snapshotPool {
return &snapshotPool{
stateProvider: stateProvider,
snapshots: make(map[snapshotKey]*snapshot),
snapshotPeers: make(map[snapshotKey]map[types.NodeID]types.NodeID),
snapshotPeers: make(map[snapshotKey]map[p2p.NodeID]p2p.NodeID),
formatIndex: make(map[uint32]map[snapshotKey]bool),
heightIndex: make(map[uint64]map[snapshotKey]bool),
peerIndex: make(map[types.NodeID]map[snapshotKey]bool),
peerIndex: make(map[p2p.NodeID]map[snapshotKey]bool),
formatBlacklist: make(map[uint32]bool),
peerBlacklist: make(map[types.NodeID]bool),
peerBlacklist: make(map[p2p.NodeID]bool),
snapshotBlacklist: make(map[snapshotKey]bool),
}
}
@@ -79,7 +79,7 @@ func newSnapshotPool(stateProvider StateProvider) *snapshotPool {
// snapshots. It 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(peerID types.NodeID, snapshot *snapshot) (bool, error) {
func (p *snapshotPool) Add(peerID p2p.NodeID, snapshot *snapshot) (bool, error) {
ctx, cancel := context.WithTimeout(context.TODO(), 30*time.Second)
defer cancel()
@@ -105,7 +105,7 @@ func (p *snapshotPool) Add(peerID types.NodeID, snapshot *snapshot) (bool, error
}
if p.snapshotPeers[key] == nil {
p.snapshotPeers[key] = make(map[types.NodeID]types.NodeID)
p.snapshotPeers[key] = make(map[p2p.NodeID]p2p.NodeID)
}
p.snapshotPeers[key][peerID] = peerID
@@ -142,7 +142,7 @@ func (p *snapshotPool) Best() *snapshot {
}
// GetPeer returns a random peer for a snapshot, if any.
func (p *snapshotPool) GetPeer(snapshot *snapshot) types.NodeID {
func (p *snapshotPool) GetPeer(snapshot *snapshot) p2p.NodeID {
peers := p.GetPeers(snapshot)
if len(peers) == 0 {
return ""
@@ -151,13 +151,13 @@ func (p *snapshotPool) GetPeer(snapshot *snapshot) types.NodeID {
}
// GetPeers returns the peers for a snapshot.
func (p *snapshotPool) GetPeers(snapshot *snapshot) []types.NodeID {
func (p *snapshotPool) GetPeers(snapshot *snapshot) []p2p.NodeID {
key := snapshot.Key()
p.Lock()
defer p.Unlock()
peers := make([]types.NodeID, 0, len(p.snapshotPeers[key]))
peers := make([]p2p.NodeID, 0, len(p.snapshotPeers[key]))
for _, peer := range p.snapshotPeers[key] {
peers = append(peers, peer)
}
@@ -254,7 +254,7 @@ func (p *snapshotPool) RejectFormat(format uint32) {
}
// RejectPeer rejects a peer. It will never be used again.
func (p *snapshotPool) RejectPeer(peerID types.NodeID) {
func (p *snapshotPool) RejectPeer(peerID p2p.NodeID) {
if len(peerID) == 0 {
return
}
@@ -267,14 +267,14 @@ func (p *snapshotPool) RejectPeer(peerID types.NodeID) {
}
// RemovePeer removes a peer from the pool, and any snapshots that no longer have peers.
func (p *snapshotPool) RemovePeer(peerID types.NodeID) {
func (p *snapshotPool) RemovePeer(peerID p2p.NodeID) {
p.Lock()
defer p.Unlock()
p.removePeer(peerID)
}
// removePeer removes a peer. The caller must hold the mutex lock.
func (p *snapshotPool) removePeer(peerID types.NodeID) {
func (p *snapshotPool) removePeer(peerID p2p.NodeID) {
for key := range p.peerIndex[peerID] {
delete(p.snapshotPeers[key], peerID)
if len(p.snapshotPeers[key]) == 0 {
internal/statesync/snapshots_test.go
+19 -19
View File
@@ -7,7 +7,7 @@ import (
"github.com/stretchr/testify/require"
"github.com/tendermint/tendermint/internal/statesync/mocks"
"github.com/tendermint/tendermint/types"
"github.com/tendermint/tendermint/pkg/p2p"
)
func TestSnapshot_Key(t *testing.T) {
@@ -42,7 +42,7 @@ func TestSnapshotPool_Add(t *testing.T) {
stateProvider := &mocks.StateProvider{}
stateProvider.On("AppHash", mock.Anything, uint64(1)).Return([]byte("app_hash"), nil)
peerID := types.NodeID("aa")
peerID := p2p.NodeID("aa")
// Adding to the pool should work
pool := newSnapshotPool(stateProvider)
@@ -56,7 +56,7 @@ func TestSnapshotPool_Add(t *testing.T) {
require.True(t, added)
// Adding again from a different peer should return false
otherNodeID := types.NodeID("bb")
otherNodeID := p2p.NodeID("bb")
added, err = pool.Add(otherNodeID, &snapshot{
Height: 1,
Format: 1,
@@ -81,8 +81,8 @@ func TestSnapshotPool_GetPeer(t *testing.T) {
s := &snapshot{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1}}
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
_, err := pool.Add(peerAID, s)
require.NoError(t, err)
@@ -118,8 +118,8 @@ func TestSnapshotPool_GetPeers(t *testing.T) {
s := &snapshot{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1}}
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
_, err := pool.Add(peerAID, s)
require.NoError(t, err)
@@ -146,13 +146,13 @@ func TestSnapshotPool_Ranked_Best(t *testing.T) {
// tie-breaker.
expectSnapshots := []struct {
snapshot *snapshot
peers []types.NodeID
peers []p2p.NodeID
}{
{&snapshot{Height: 2, Format: 2, Chunks: 4, Hash: []byte{1, 3}}, []types.NodeID{"AA", "BB", "CC", "DD"}},
{&snapshot{Height: 1, Format: 1, Chunks: 4, Hash: []byte{1, 2}}, []types.NodeID{"AA", "BB", "CC", "DD"}},
{&snapshot{Height: 2, Format: 2, Chunks: 5, Hash: []byte{1, 2}}, []types.NodeID{"AA", "BB", "CC"}},
{&snapshot{Height: 2, Format: 1, Chunks: 3, Hash: []byte{1, 2}}, []types.NodeID{"AA", "BB", "CC"}},
{&snapshot{Height: 1, Format: 2, Chunks: 5, Hash: []byte{1, 2}}, []types.NodeID{"AA", "BB", "CC"}},
{&snapshot{Height: 2, Format: 2, Chunks: 4, Hash: []byte{1, 3}}, []p2p.NodeID{"AA", "BB", "CC", "DD"}},
{&snapshot{Height: 1, Format: 1, Chunks: 4, Hash: []byte{1, 2}}, []p2p.NodeID{"AA", "BB", "CC", "DD"}},
{&snapshot{Height: 2, Format: 2, Chunks: 5, Hash: []byte{1, 2}}, []p2p.NodeID{"AA", "BB", "CC"}},
{&snapshot{Height: 2, Format: 1, Chunks: 3, Hash: []byte{1, 2}}, []p2p.NodeID{"AA", "BB", "CC"}},
{&snapshot{Height: 1, Format: 2, Chunks: 5, Hash: []byte{1, 2}}, []p2p.NodeID{"AA", "BB", "CC"}},
}
// Add snapshots in reverse order, to make sure the pool enforces some order.
@@ -186,7 +186,7 @@ func TestSnapshotPool_Reject(t *testing.T) {
stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
pool := newSnapshotPool(stateProvider)
peerID := types.NodeID("aa")
peerID := p2p.NodeID("aa")
snapshots := []*snapshot{
{Height: 2, Format: 2, Chunks: 1, Hash: []byte{1, 2}},
@@ -216,7 +216,7 @@ func TestSnapshotPool_RejectFormat(t *testing.T) {
stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
pool := newSnapshotPool(stateProvider)
peerID := types.NodeID("aa")
peerID := p2p.NodeID("aa")
snapshots := []*snapshot{
{Height: 2, Format: 2, Chunks: 1, Hash: []byte{1, 2}},
@@ -247,8 +247,8 @@ func TestSnapshotPool_RejectPeer(t *testing.T) {
stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
pool := newSnapshotPool(stateProvider)
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
s1 := &snapshot{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1}}
s2 := &snapshot{Height: 2, Format: 1, Chunks: 1, Hash: []byte{2}}
@@ -289,8 +289,8 @@ func TestSnapshotPool_RemovePeer(t *testing.T) {
stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
pool := newSnapshotPool(stateProvider)
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
s1 := &snapshot{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1}}
s2 := &snapshot{Height: 2, Format: 1, Chunks: 1, Hash: []byte{2}}
internal/statesync/stateprovider.go
+3 -3
View File
@@ -15,9 +15,9 @@ import (
lighthttp "github.com/tendermint/tendermint/light/provider/http"
lightrpc "github.com/tendermint/tendermint/light/rpc"
lightdb "github.com/tendermint/tendermint/light/store/db"
"github.com/tendermint/tendermint/pkg/metadata"
rpchttp "github.com/tendermint/tendermint/rpc/client/http"
sm "github.com/tendermint/tendermint/state"
"github.com/tendermint/tendermint/types"
)
//go:generate ../../scripts/mockery_generate.sh StateProvider
@@ -28,7 +28,7 @@ type StateProvider interface {
// AppHash returns the app hash after the given height has been committed.
AppHash(ctx context.Context, height uint64) ([]byte, error)
// Commit returns the commit at the given height.
Commit(ctx context.Context, height uint64) (*types.Commit, error)
Commit(ctx context.Context, height uint64) (*metadata.Commit, error)
// State returns a state object at the given height.
State(ctx context.Context, height uint64) (sm.State, error)
}
@@ -148,7 +148,7 @@ func (s *lightClientStateProvider) AppHash(ctx context.Context, height uint64) (
}
// Commit implements StateProvider.
func (s *lightClientStateProvider) Commit(ctx context.Context, height uint64) (*types.Commit, error) {
func (s *lightClientStateProvider) Commit(ctx context.Context, height uint64) (*metadata.Commit, error) {
s.Lock()
defer s.Unlock()
header, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height), time.Now())
internal/statesync/syncer.go
+9 -8
View File
@@ -7,15 +7,16 @@ import (
"fmt"
"time"
abci "github.com/tendermint/tendermint/abci/types"
"github.com/tendermint/tendermint/config"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/internal/p2p"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/pkg/abci"
"github.com/tendermint/tendermint/pkg/metadata"
p2ptypes "github.com/tendermint/tendermint/pkg/p2p"
ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
"github.com/tendermint/tendermint/proxy"
sm "github.com/tendermint/tendermint/state"
"github.com/tendermint/tendermint/types"
)
const (
@@ -117,7 +118,7 @@ func (s *syncer) AddChunk(chunk *chunk) (bool, error) {
// AddSnapshot adds a snapshot to the snapshot pool. It returns true if a new, previously unseen
// snapshot was accepted and added.
func (s *syncer) AddSnapshot(peerID types.NodeID, snapshot *snapshot) (bool, error) {
func (s *syncer) AddSnapshot(peerID p2ptypes.NodeID, snapshot *snapshot) (bool, error) {
added, err := s.snapshots.Add(peerID, snapshot)
if err != nil {
return false, err
@@ -131,7 +132,7 @@ func (s *syncer) AddSnapshot(peerID types.NodeID, snapshot *snapshot) (bool, err
// AddPeer adds a peer to the pool. For now we just keep it simple and send a
// single request to discover snapshots, later we may want to do retries and stuff.
func (s *syncer) AddPeer(peerID types.NodeID) {
func (s *syncer) AddPeer(peerID p2ptypes.NodeID) {
s.logger.Debug("Requesting snapshots from peer", "peer", peerID)
s.snapshotCh <- p2p.Envelope{
To: peerID,
@@ -140,7 +141,7 @@ func (s *syncer) AddPeer(peerID types.NodeID) {
}
// RemovePeer removes a peer from the pool.
func (s *syncer) RemovePeer(peerID types.NodeID) {
func (s *syncer) RemovePeer(peerID p2ptypes.NodeID) {
s.logger.Debug("Removing peer from sync", "peer", peerID)
s.snapshots.RemovePeer(peerID)
}
@@ -152,7 +153,7 @@ func (s *syncer) SyncAny(
ctx context.Context,
discoveryTime time.Duration,
requestSnapshots func(),
) (sm.State, *types.Commit, error) {
) (sm.State, *metadata.Commit, error) {
if discoveryTime != 0 && discoveryTime < minimumDiscoveryTime {
discoveryTime = minimumDiscoveryTime
@@ -250,7 +251,7 @@ func (s *syncer) SyncAny(
// Sync executes a sync for a specific snapshot, returning the latest state and block commit which
// the caller must use to bootstrap the node.
func (s *syncer) Sync(ctx context.Context, snapshot *snapshot, chunks *chunkQueue) (sm.State, *types.Commit, error) {
func (s *syncer) Sync(ctx context.Context, snapshot *snapshot, chunks *chunkQueue) (sm.State, *metadata.Commit, error) {
s.mtx.Lock()
if s.chunks != nil {
s.mtx.Unlock()
@@ -389,7 +390,7 @@ func (s *syncer) applyChunks(ctx context.Context, chunks *chunkQueue) error {
// Reject any senders as requested by the app
for _, sender := range resp.RejectSenders {
if sender != "" {
peerID := types.NodeID(sender)
peerID := p2ptypes.NodeID(sender)
s.snapshots.RejectPeer(peerID)
if err := chunks.DiscardSender(peerID); err != nil {
internal/statesync/syncer_test.go
+23 -21
View File
@@ -11,14 +11,16 @@ import (
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
abci "github.com/tendermint/tendermint/abci/types"
tmsync "github.com/tendermint/tendermint/internal/libs/sync"
"github.com/tendermint/tendermint/internal/statesync/mocks"
"github.com/tendermint/tendermint/pkg/abci"
"github.com/tendermint/tendermint/pkg/consensus"
"github.com/tendermint/tendermint/pkg/metadata"
"github.com/tendermint/tendermint/pkg/p2p"
ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
"github.com/tendermint/tendermint/proxy"
proxymocks "github.com/tendermint/tendermint/proxy/mocks"
sm "github.com/tendermint/tendermint/state"
"github.com/tendermint/tendermint/types"
"github.com/tendermint/tendermint/version"
)
@@ -36,19 +38,19 @@ func TestSyncer_SyncAny(t *testing.T) {
},
LastBlockHeight: 1,
LastBlockID: types.BlockID{Hash: []byte("blockhash")},
LastBlockID: metadata.BlockID{Hash: []byte("blockhash")},
LastBlockTime: time.Now(),
LastResultsHash: []byte("last_results_hash"),
AppHash: []byte("app_hash"),
LastValidators: &types.ValidatorSet{Proposer: &types.Validator{Address: []byte("val1")}},
Validators: &types.ValidatorSet{Proposer: &types.Validator{Address: []byte("val2")}},
NextValidators: &types.ValidatorSet{Proposer: &types.Validator{Address: []byte("val3")}},
LastValidators: &consensus.ValidatorSet{Proposer: &consensus.Validator{Address: []byte("val1")}},
Validators: &consensus.ValidatorSet{Proposer: &consensus.Validator{Address: []byte("val2")}},
NextValidators: &consensus.ValidatorSet{Proposer: &consensus.Validator{Address: []byte("val3")}},
ConsensusParams: *types.DefaultConsensusParams(),
ConsensusParams: *consensus.DefaultConsensusParams(),
LastHeightConsensusParamsChanged: 1,
}
commit := &types.Commit{BlockID: types.BlockID{Hash: []byte("blockhash")}}
commit := &metadata.Commit{BlockID: metadata.BlockID{Hash: []byte("blockhash")}}
chunks := []*chunk{
{Height: 1, Format: 1, Index: 0, Chunk: []byte{1, 1, 0}},
@@ -65,9 +67,9 @@ func TestSyncer_SyncAny(t *testing.T) {
connSnapshot := &proxymocks.AppConnSnapshot{}
connQuery := &proxymocks.AppConnQuery{}
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerCID := types.NodeID("cc")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
peerCID := p2p.NodeID("cc")
rts := setup(t, connSnapshot, connQuery, stateProvider, 3)
// Adding a chunk should error when no sync is in progress
@@ -216,7 +218,7 @@ func TestSyncer_SyncAny_abort(t *testing.T) {
rts := setup(t, nil, nil, stateProvider, 2)
s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
peerID := types.NodeID("aa")
peerID := p2p.NodeID("aa")
_, err := rts.syncer.AddSnapshot(peerID, s)
require.NoError(t, err)
@@ -241,7 +243,7 @@ func TestSyncer_SyncAny_reject(t *testing.T) {
s12 := &snapshot{Height: 1, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
s11 := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
peerID := types.NodeID("aa")
peerID := p2p.NodeID("aa")
_, err := rts.syncer.AddSnapshot(peerID, s22)
require.NoError(t, err)
@@ -280,7 +282,7 @@ func TestSyncer_SyncAny_reject_format(t *testing.T) {
s12 := &snapshot{Height: 1, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
s11 := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
peerID := types.NodeID("aa")
peerID := p2p.NodeID("aa")
_, err := rts.syncer.AddSnapshot(peerID, s22)
require.NoError(t, err)
@@ -310,9 +312,9 @@ func TestSyncer_SyncAny_reject_sender(t *testing.T) {
rts := setup(t, nil, nil, stateProvider, 2)
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerCID := types.NodeID("cc")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
peerCID := p2p.NodeID("cc")
// sbc will be offered first, which will be rejected with reject_sender, causing all snapshots
// submitted by both b and c (i.e. sb, sc, sbc) to be rejected. Finally, sa will reject and
@@ -359,7 +361,7 @@ func TestSyncer_SyncAny_abciError(t *testing.T) {
errBoom := errors.New("boom")
s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
peerID := types.NodeID("aa")
peerID := p2p.NodeID("aa")
_, err := rts.syncer.AddSnapshot(peerID, s)
require.NoError(t, err)
@@ -560,9 +562,9 @@ func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
// Set up three peers across two snapshots, and ask for one of them to be banned.
// It should be banned from all snapshots.
peerAID := types.NodeID("aa")
peerBID := types.NodeID("bb")
peerCID := types.NodeID("cc")
peerAID := p2p.NodeID("aa")
peerBID := p2p.NodeID("bb")
peerCID := p2p.NodeID("cc")
s1 := &snapshot{Height: 1, Format: 1, Chunks: 3}
s2 := &snapshot{Height: 2, Format: 1, Chunks: 3}