p2p: renames for reactors and routing layer internal moves (#6547)

internal/statesync/chunks.go

@@ -0,0 +1,357 @@
+package statesync
+
+import (
+	"errors"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strconv"
+	"time"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+	"github.com/tendermint/tendermint/internal/p2p"
+)
+
+// errDone is returned by chunkQueue.Next() when all chunks have been returned.
+var errDone = errors.New("chunk queue has completed")
+
+// chunk contains data for a chunk.
+type chunk struct {
+	Height uint64
+	Format uint32
+	Index  uint32
+	Chunk  []byte
+	Sender p2p.NodeID
+}
+
+// chunkQueue manages chunks for a state sync process, ordering them if requested. It acts as an
+// iterator over all chunks, but callers can request chunks to be retried, optionally after
+// refetching.
+type chunkQueue struct {
+	tmsync.Mutex
+	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]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
+}
+
+// newChunkQueue creates a new chunk queue for a snapshot, using a temp dir for storage.
+// Callers must call Close() when done.
+func newChunkQueue(snapshot *snapshot, tempDir string) (*chunkQueue, error) {
+	dir, err := ioutil.TempDir(tempDir, "tm-statesync")
+	if err != nil {
+		return nil, fmt.Errorf("unable to create temp dir for state sync chunks: %w", err)
+	}
+	if snapshot.Chunks == 0 {
+		return nil, errors.New("snapshot has no chunks")
+	}
+
+	return &chunkQueue{
+		snapshot:       snapshot,
+		dir:            dir,
+		chunkFiles:     make(map[uint32]string, 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),
+	}, nil
+}
+
+// Add adds a chunk to the queue. It ignores chunks that already exist, returning false.
+func (q *chunkQueue) Add(chunk *chunk) (bool, error) {
+	if chunk == nil || chunk.Chunk == nil {
+		return false, errors.New("cannot add nil chunk")
+	}
+
+	q.Lock()
+	defer q.Unlock()
+
+	if q.snapshot == nil {
+		return false, nil // queue is closed
+	}
+	if chunk.Height != q.snapshot.Height {
+		return false, fmt.Errorf("invalid chunk height %v, expected %v", chunk.Height, q.snapshot.Height)
+	}
+	if chunk.Format != q.snapshot.Format {
+		return false, fmt.Errorf("invalid chunk format %v, expected %v", chunk.Format, q.snapshot.Format)
+	}
+	if chunk.Index >= q.snapshot.Chunks {
+		return false, fmt.Errorf("received unexpected chunk %v", chunk.Index)
+	}
+	if q.chunkFiles[chunk.Index] != "" {
+		return false, nil
+	}
+
+	path := filepath.Join(q.dir, strconv.FormatUint(uint64(chunk.Index), 10))
+	err := ioutil.WriteFile(path, chunk.Chunk, 0600)
+	if err != nil {
+		return false, fmt.Errorf("failed to save chunk %v to file %v: %w", chunk.Index, path, err)
+	}
+
+	q.chunkFiles[chunk.Index] = path
+	q.chunkSenders[chunk.Index] = chunk.Sender
+
+	// Signal any waiters that the chunk has arrived.
+	for _, waiter := range q.waiters[chunk.Index] {
+		waiter <- chunk.Index
+		close(waiter)
+	}
+
+	delete(q.waiters, chunk.Index)
+
+	return true, nil
+}
+
+// Allocate allocates a chunk to the caller, making it responsible for fetching it. Returns
+// errDone once no chunks are left or the queue is closed.
+func (q *chunkQueue) Allocate() (uint32, error) {
+	q.Lock()
+	defer q.Unlock()
+
+	if q.snapshot == nil {
+		return 0, errDone
+	}
+
+	if uint32(len(q.chunkAllocated)) >= q.snapshot.Chunks {
+		return 0, errDone
+	}
+
+	for i := uint32(0); i < q.snapshot.Chunks; i++ {
+		if !q.chunkAllocated[i] {
+			q.chunkAllocated[i] = true
+			return i, nil
+		}
+	}
+
+	return 0, errDone
+}
+
+// Close closes the chunk queue, cleaning up all temporary files.
+func (q *chunkQueue) Close() error {
+	q.Lock()
+	defer q.Unlock()
+
+	if q.snapshot == nil {
+		return nil
+	}
+
+	for _, waiters := range q.waiters {
+		for _, waiter := range waiters {
+			close(waiter)
+		}
+	}
+
+	q.waiters = nil
+	q.snapshot = nil
+
+	if err := os.RemoveAll(q.dir); err != nil {
+		return fmt.Errorf("failed to clean up state sync tempdir %v: %w", q.dir, err)
+	}
+
+	return nil
+}
+
+// Discard discards a chunk. It will be removed from the queue, available for allocation, and can
+// be added and returned via Next() again. If the chunk is not already in the queue this does
+// nothing, to avoid it being allocated to multiple fetchers.
+func (q *chunkQueue) Discard(index uint32) error {
+	q.Lock()
+	defer q.Unlock()
+	return q.discard(index)
+}
+
+// discard discards a chunk, scheduling it for refetching. The caller must hold the mutex lock.
+func (q *chunkQueue) discard(index uint32) error {
+	if q.snapshot == nil {
+		return nil
+	}
+
+	path := q.chunkFiles[index]
+	if path == "" {
+		return nil
+	}
+
+	if err := os.Remove(path); err != nil {
+		return fmt.Errorf("failed to remove chunk %v: %w", index, err)
+	}
+
+	delete(q.chunkFiles, index)
+	delete(q.chunkReturned, index)
+	delete(q.chunkAllocated, index)
+
+	return nil
+}
+
+// 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 p2p.NodeID) error {
+	q.Lock()
+	defer q.Unlock()
+
+	for index, sender := range q.chunkSenders {
+		if sender == peerID && !q.chunkReturned[index] {
+			err := q.discard(index)
+			if err != nil {
+				return err
+			}
+
+			delete(q.chunkSenders, index)
+		}
+	}
+
+	return nil
+}
+
+// GetSender returns the sender of the chunk with the given index, or empty if
+// not found.
+func (q *chunkQueue) GetSender(index uint32) p2p.NodeID {
+	q.Lock()
+	defer q.Unlock()
+	return q.chunkSenders[index]
+}
+
+// Has checks whether a chunk exists in the queue.
+func (q *chunkQueue) Has(index uint32) bool {
+	q.Lock()
+	defer q.Unlock()
+	return q.chunkFiles[index] != ""
+}
+
+// load loads a chunk from disk, or nil if the chunk is not in the queue. The caller must hold the
+// mutex lock.
+func (q *chunkQueue) load(index uint32) (*chunk, error) {
+	path, ok := q.chunkFiles[index]
+	if !ok {
+		return nil, nil
+	}
+
+	body, err := ioutil.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("failed to load chunk %v: %w", index, err)
+	}
+
+	return &chunk{
+		Height: q.snapshot.Height,
+		Format: q.snapshot.Format,
+		Index:  index,
+		Chunk:  body,
+		Sender: q.chunkSenders[index],
+	}, nil
+}
+
+// Next returns the next chunk from the queue, or errDone if all chunks have been returned. It
+// blocks until the chunk is available. Concurrent Next() calls may return the same chunk.
+func (q *chunkQueue) Next() (*chunk, error) {
+	q.Lock()
+
+	var chunk *chunk
+	index, err := q.nextUp()
+	if err == nil {
+		chunk, err = q.load(index)
+		if err == nil {
+			q.chunkReturned[index] = true
+		}
+	}
+
+	q.Unlock()
+
+	if chunk != nil || err != nil {
+		return chunk, err
+	}
+
+	select {
+	case _, ok := <-q.WaitFor(index):
+		if !ok {
+			return nil, errDone // queue closed
+		}
+	case <-time.After(chunkTimeout):
+		return nil, errTimeout
+	}
+
+	q.Lock()
+	defer q.Unlock()
+
+	chunk, err = q.load(index)
+	if err != nil {
+		return nil, err
+	}
+
+	q.chunkReturned[index] = true
+	return chunk, nil
+}
+
+// nextUp returns the next chunk to be returned, or errDone if all chunks have been returned. The
+// caller must hold the mutex lock.
+func (q *chunkQueue) nextUp() (uint32, error) {
+	if q.snapshot == nil {
+		return 0, errDone
+	}
+
+	for i := uint32(0); i < q.snapshot.Chunks; i++ {
+		if !q.chunkReturned[i] {
+			return i, nil
+		}
+	}
+
+	return 0, errDone
+}
+
+// Retry schedules a chunk to be retried, without refetching it.
+func (q *chunkQueue) Retry(index uint32) {
+	q.Lock()
+	defer q.Unlock()
+	delete(q.chunkReturned, index)
+}
+
+// RetryAll schedules all chunks to be retried, without refetching them.
+func (q *chunkQueue) RetryAll() {
+	q.Lock()
+	defer q.Unlock()
+	q.chunkReturned = make(map[uint32]bool)
+}
+
+// Size returns the total number of chunks for the snapshot and queue, or 0 when closed.
+func (q *chunkQueue) Size() uint32 {
+	q.Lock()
+	defer q.Unlock()
+
+	if q.snapshot == nil {
+		return 0
+	}
+
+	return q.snapshot.Chunks
+}
+
+// WaitFor returns a channel that receives a chunk index when it arrives in the queue, or
+// immediately if it has already arrived. The channel is closed without a value if the queue is
+// closed or if the chunk index is not valid.
+func (q *chunkQueue) WaitFor(index uint32) <-chan uint32 {
+	q.Lock()
+	defer q.Unlock()
+
+	ch := make(chan uint32, 1)
+	switch {
+	case q.snapshot == nil:
+		close(ch)
+
+	case index >= q.snapshot.Chunks:
+		close(ch)
+
+	case q.chunkFiles[index] != "":
+		ch <- index
+		close(ch)
+
+	default:
+		if q.waiters[index] == nil {
+			q.waiters[index] = make([]chan<- uint32, 0)
+		}
+
+		q.waiters[index] = append(q.waiters[index], ch)
+	}
+
+	return ch
+}

internal/statesync/chunks_test.go

@@ -0,0 +1,554 @@
+package statesync
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/tendermint/tendermint/internal/p2p"
+)
+
+func setupChunkQueue(t *testing.T) (*chunkQueue, func()) {
+	snapshot := &snapshot{
+		Height:   3,
+		Format:   1,
+		Chunks:   5,
+		Hash:     []byte{7},
+		Metadata: nil,
+	}
+	queue, err := newChunkQueue(snapshot, "")
+	require.NoError(t, err)
+	teardown := func() {
+		err := queue.Close()
+		require.NoError(t, err)
+	}
+	return queue, teardown
+}
+
+func TestNewChunkQueue_TempDir(t *testing.T) {
+	snapshot := &snapshot{
+		Height:   3,
+		Format:   1,
+		Chunks:   5,
+		Hash:     []byte{7},
+		Metadata: nil,
+	}
+	dir, err := ioutil.TempDir("", "newchunkqueue")
+	require.NoError(t, err)
+	defer os.RemoveAll(dir)
+	queue, err := newChunkQueue(snapshot, dir)
+	require.NoError(t, err)
+
+	files, err := ioutil.ReadDir(dir)
+	require.NoError(t, err)
+	assert.Len(t, files, 1)
+
+	err = queue.Close()
+	require.NoError(t, err)
+
+	files, err = ioutil.ReadDir(dir)
+	require.NoError(t, err)
+	assert.Len(t, files, 0)
+}
+
+func TestChunkQueue(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	// Adding the first chunk should be fine
+	added, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}})
+	require.NoError(t, err)
+	assert.True(t, added)
+
+	// Adding the last chunk should also be fine
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 4, Chunk: []byte{3, 1, 4}})
+	require.NoError(t, err)
+	assert.True(t, added)
+
+	// Adding the first or last chunks again should return false
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}})
+	require.NoError(t, err)
+	assert.False(t, added)
+
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 4, Chunk: []byte{3, 1, 4}})
+	require.NoError(t, err)
+	assert.False(t, added)
+
+	// Adding the remaining chunks in reverse should be fine
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 3, Chunk: []byte{3, 1, 3}})
+	require.NoError(t, err)
+	assert.True(t, added)
+
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 2, Chunk: []byte{3, 1, 2}})
+	require.NoError(t, err)
+	assert.True(t, added)
+
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{3, 1, 1}})
+	require.NoError(t, err)
+	assert.True(t, added)
+
+	// At this point, we should be able to retrieve them all via Next
+	for i := 0; i < 5; i++ {
+		c, err := queue.Next()
+		require.NoError(t, err)
+		assert.Equal(t, &chunk{Height: 3, Format: 1, Index: uint32(i), Chunk: []byte{3, 1, byte(i)}}, c)
+	}
+	_, err = queue.Next()
+	require.Error(t, err)
+	assert.Equal(t, errDone, err)
+
+	// It should still be possible to try to add chunks (which will be ignored)
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}})
+	require.NoError(t, err)
+	assert.False(t, added)
+
+	// After closing the queue it will also return false
+	err = queue.Close()
+	require.NoError(t, err)
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}})
+	require.NoError(t, err)
+	assert.False(t, added)
+
+	// Closing the queue again should also be fine
+	err = queue.Close()
+	require.NoError(t, err)
+}
+
+func TestChunkQueue_Add_ChunkErrors(t *testing.T) {
+	testcases := map[string]struct {
+		chunk *chunk
+	}{
+		"nil chunk":     {nil},
+		"nil body":      {&chunk{Height: 3, Format: 1, Index: 0, Chunk: nil}},
+		"wrong height":  {&chunk{Height: 9, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}}},
+		"wrong format":  {&chunk{Height: 3, Format: 9, Index: 0, Chunk: []byte{3, 1, 0}}},
+		"invalid index": {&chunk{Height: 3, Format: 1, Index: 5, Chunk: []byte{3, 1, 0}}},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			queue, teardown := setupChunkQueue(t)
+			defer teardown()
+			_, err := queue.Add(tc.chunk)
+			require.Error(t, err)
+		})
+	}
+}
+
+func TestChunkQueue_Allocate(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	for i := uint32(0); i < queue.Size(); i++ {
+		index, err := queue.Allocate()
+		require.NoError(t, err)
+		assert.EqualValues(t, i, index)
+	}
+
+	_, err := queue.Allocate()
+	require.Error(t, err)
+	assert.Equal(t, errDone, err)
+
+	for i := uint32(0); i < queue.Size(); i++ {
+		_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: i, Chunk: []byte{byte(i)}})
+		require.NoError(t, err)
+	}
+
+	// After all chunks have been allocated and retrieved, discarding a chunk will reallocate it.
+	err = queue.Discard(2)
+	require.NoError(t, err)
+
+	index, err := queue.Allocate()
+	require.NoError(t, err)
+	assert.EqualValues(t, 2, index)
+	_, err = queue.Allocate()
+	require.Error(t, err)
+	assert.Equal(t, errDone, err)
+
+	// Discarding a chunk the closing the queue will return errDone.
+	err = queue.Discard(2)
+	require.NoError(t, err)
+	err = queue.Close()
+	require.NoError(t, err)
+	_, err = queue.Allocate()
+	require.Error(t, err)
+	assert.Equal(t, errDone, err)
+}
+
+func TestChunkQueue_Discard(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	// Add a few chunks to the queue and fetch a couple
+	_, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{byte(0)}})
+	require.NoError(t, err)
+	_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{byte(1)}})
+	require.NoError(t, err)
+	_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 2, Chunk: []byte{byte(2)}})
+	require.NoError(t, err)
+
+	c, err := queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 0, c.Index)
+	c, err = queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 1, c.Index)
+
+	// Discarding the first chunk and re-adding it should cause it to be returned
+	// immediately by Next(), before procceeding with chunk 2
+	err = queue.Discard(0)
+	require.NoError(t, err)
+	added, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{byte(0)}})
+	require.NoError(t, err)
+	assert.True(t, added)
+	c, err = queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 0, c.Index)
+	c, err = queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 2, c.Index)
+
+	// Discard then allocate, add and fetch all chunks
+	for i := uint32(0); i < queue.Size(); i++ {
+		err := queue.Discard(i)
+		require.NoError(t, err)
+	}
+	for i := uint32(0); i < queue.Size(); i++ {
+		_, err := queue.Allocate()
+		require.NoError(t, err)
+		_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: i, Chunk: []byte{byte(i)}})
+		require.NoError(t, err)
+		c, err = queue.Next()
+		require.NoError(t, err)
+		assert.EqualValues(t, i, c.Index)
+	}
+
+	// Discarding a non-existent chunk does nothing.
+	err = queue.Discard(99)
+	require.NoError(t, err)
+
+	// When discard a couple of chunks, we should be able to allocate, add, and fetch them again.
+	err = queue.Discard(3)
+	require.NoError(t, err)
+	err = queue.Discard(1)
+	require.NoError(t, err)
+
+	index, err := queue.Allocate()
+	require.NoError(t, err)
+	assert.EqualValues(t, 1, index)
+	index, err = queue.Allocate()
+	require.NoError(t, err)
+	assert.EqualValues(t, 3, index)
+
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 3, Chunk: []byte{3}})
+	require.NoError(t, err)
+	assert.True(t, added)
+	added, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{1}})
+	require.NoError(t, err)
+	assert.True(t, added)
+
+	chunk, err := queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 1, chunk.Index)
+
+	chunk, err = queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 3, chunk.Index)
+
+	_, err = queue.Next()
+	require.Error(t, err)
+	assert.Equal(t, errDone, err)
+
+	// After closing the queue, discarding does nothing
+	err = queue.Close()
+	require.NoError(t, err)
+	err = queue.Discard(2)
+	require.NoError(t, err)
+}
+
+func TestChunkQueue_DiscardSender(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	// Allocate and add all chunks to the queue
+	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)
+		_, err = queue.Add(&chunk{
+			Height: 3,
+			Format: 1,
+			Index:  i,
+			Chunk:  []byte{byte(i)},
+			Sender: senders[int(i)%len(senders)],
+		})
+		require.NoError(t, err)
+	}
+
+	// Fetch the first three chunks
+	for i := uint32(0); i < 3; i++ {
+		_, err := queue.Next()
+		require.NoError(t, err)
+	}
+
+	// Discarding an unknown sender should do nothing
+	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(p2p.NodeID("b"))
+	require.NoError(t, err)
+	index, err := queue.Allocate()
+	require.NoError(t, err)
+	assert.EqualValues(t, 4, index)
+	_, err = queue.Allocate()
+	assert.Equal(t, errDone, err)
+}
+
+func TestChunkQueue_GetSender(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	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)
+	_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{2}, Sender: peerBID})
+	require.NoError(t, err)
+
+	assert.EqualValues(t, "aa", queue.GetSender(0))
+	assert.EqualValues(t, "bb", queue.GetSender(1))
+	assert.EqualValues(t, "", queue.GetSender(2))
+
+	// After the chunk has been processed, we should still know who the sender was
+	chunk, err := queue.Next()
+	require.NoError(t, err)
+	require.NotNil(t, chunk)
+	require.EqualValues(t, 0, chunk.Index)
+	assert.EqualValues(t, "aa", queue.GetSender(0))
+}
+
+func TestChunkQueue_Next(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	// Next should block waiting for the next chunks, even when given out of order.
+	chNext := make(chan *chunk, 10)
+	go func() {
+		for {
+			c, err := queue.Next()
+			if err == errDone {
+				close(chNext)
+				break
+			}
+			require.NoError(t, err)
+			chNext <- c
+		}
+	}()
+
+	assert.Empty(t, chNext)
+	_, 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:
+		assert.Fail(t, "channel should be empty")
+	default:
+	}
+
+	_, 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: p2p.NodeID("a")},
+		<-chNext)
+	assert.Equal(t,
+		&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: p2p.NodeID("e")})
+	require.NoError(t, err)
+	select {
+	case <-chNext:
+		assert.Fail(t, "channel should be empty")
+	default:
+	}
+
+	_, 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: p2p.NodeID("d")})
+	require.NoError(t, err)
+
+	assert.Equal(t,
+		&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: p2p.NodeID("d")},
+		<-chNext)
+	assert.Equal(t,
+		&chunk{Height: 3, Format: 1, Index: 4, Chunk: []byte{3, 1, 4}, Sender: p2p.NodeID("e")},
+		<-chNext)
+
+	_, ok := <-chNext
+	assert.False(t, ok, "channel should be closed")
+
+	// Calling next on a finished queue should return done
+	_, err = queue.Next()
+	assert.Equal(t, errDone, err)
+}
+
+func TestChunkQueue_Next_Closed(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	// Calling Next on a closed queue should return done
+	_, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{3, 1, 1}})
+	require.NoError(t, err)
+	err = queue.Close()
+	require.NoError(t, err)
+
+	_, err = queue.Next()
+	assert.Equal(t, errDone, err)
+}
+
+func TestChunkQueue_Retry(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	// Allocate and add all chunks to the queue
+	for i := uint32(0); i < queue.Size(); i++ {
+		_, err := queue.Allocate()
+		require.NoError(t, err)
+		_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: i, Chunk: []byte{byte(i)}})
+		require.NoError(t, err)
+		_, err = queue.Next()
+		require.NoError(t, err)
+	}
+
+	// Retrying a couple of chunks makes Next() return them, but they are not allocatable
+	queue.Retry(3)
+	queue.Retry(1)
+
+	_, err := queue.Allocate()
+	assert.Equal(t, errDone, err)
+
+	chunk, err := queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 1, chunk.Index)
+
+	chunk, err = queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 3, chunk.Index)
+
+	_, err = queue.Next()
+	assert.Equal(t, errDone, err)
+}
+
+func TestChunkQueue_RetryAll(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	// Allocate and add all chunks to the queue
+	for i := uint32(0); i < queue.Size(); i++ {
+		_, err := queue.Allocate()
+		require.NoError(t, err)
+		_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: i, Chunk: []byte{byte(i)}})
+		require.NoError(t, err)
+		_, err = queue.Next()
+		require.NoError(t, err)
+	}
+
+	_, err := queue.Next()
+	assert.Equal(t, errDone, err)
+
+	queue.RetryAll()
+
+	_, err = queue.Allocate()
+	assert.Equal(t, errDone, err)
+
+	for i := uint32(0); i < queue.Size(); i++ {
+		chunk, err := queue.Next()
+		require.NoError(t, err)
+		assert.EqualValues(t, i, chunk.Index)
+	}
+
+	_, err = queue.Next()
+	assert.Equal(t, errDone, err)
+}
+
+func TestChunkQueue_Size(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	assert.EqualValues(t, 5, queue.Size())
+
+	err := queue.Close()
+	require.NoError(t, err)
+	assert.EqualValues(t, 0, queue.Size())
+}
+
+func TestChunkQueue_WaitFor(t *testing.T) {
+	queue, teardown := setupChunkQueue(t)
+	defer teardown()
+
+	waitFor1 := queue.WaitFor(1)
+	waitFor4 := queue.WaitFor(4)
+
+	// Adding 0 and 2 should not trigger waiters
+	_, err := queue.Add(&chunk{Height: 3, Format: 1, Index: 0, Chunk: []byte{3, 1, 0}})
+	require.NoError(t, err)
+	_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 2, Chunk: []byte{3, 1, 2}})
+	require.NoError(t, err)
+	select {
+	case <-waitFor1:
+		require.Fail(t, "WaitFor(1) should not trigger on 0 or 2")
+	case <-waitFor4:
+		require.Fail(t, "WaitFor(4) should not trigger on 0 or 2")
+	default:
+	}
+
+	// Adding 1 should trigger WaitFor(1), but not WaitFor(4). The channel should be closed.
+	_, err = queue.Add(&chunk{Height: 3, Format: 1, Index: 1, Chunk: []byte{3, 1, 1}})
+	require.NoError(t, err)
+	assert.EqualValues(t, 1, <-waitFor1)
+	_, ok := <-waitFor1
+	assert.False(t, ok)
+	select {
+	case <-waitFor4:
+		require.Fail(t, "WaitFor(4) should not trigger on 0 or 2")
+	default:
+	}
+
+	// Fetch the first chunk. At this point, waiting for either 0 (retrieved from pool) or 1
+	// (queued in pool) should immediately return true.
+	c, err := queue.Next()
+	require.NoError(t, err)
+	assert.EqualValues(t, 0, c.Index)
+
+	w := queue.WaitFor(0)
+	assert.EqualValues(t, 0, <-w)
+	_, ok = <-w
+	assert.False(t, ok)
+
+	w = queue.WaitFor(1)
+	assert.EqualValues(t, 1, <-w)
+	_, ok = <-w
+	assert.False(t, ok)
+
+	// Close the queue. This should cause the waiter for 4 to close, and also cause any future
+	// waiters to get closed channels.
+	err = queue.Close()
+	require.NoError(t, err)
+	_, ok = <-waitFor4
+	assert.False(t, ok)
+
+	w = queue.WaitFor(3)
+	_, ok = <-w
+	assert.False(t, ok)
+}

internal/statesync/mocks/state_provider.go

@@ -0,0 +1,84 @@
+// Code generated by mockery v2.5.1. DO NOT EDIT.
+
+package mocks
+
+import (
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+	state "github.com/tendermint/tendermint/state"
+
+	types "github.com/tendermint/tendermint/types"
+)
+
+// StateProvider is an autogenerated mock type for the StateProvider type
+type StateProvider struct {
+	mock.Mock
+}
+
+// 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(context.Context, uint64) []byte); ok {
+		r0 = rf(ctx, height)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]byte)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok {
+		r1 = rf(ctx, height)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// 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(context.Context, uint64) *types.Commit); ok {
+		r0 = rf(ctx, height)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*types.Commit)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, uint64) error); ok {
+		r1 = rf(ctx, height)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// 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(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(context.Context, uint64) error); ok {
+		r1 = rf(ctx, height)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}

internal/statesync/reactor.go

@@ -0,0 +1,498 @@
+package statesync
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"time"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+	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"
+	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
+	"github.com/tendermint/tendermint/proxy"
+	sm "github.com/tendermint/tendermint/state"
+	"github.com/tendermint/tendermint/types"
+)
+
+var (
+	_ service.Service = (*Reactor)(nil)
+	_ p2p.Wrapper     = (*ssproto.Message)(nil)
+
+	// ChannelShims contains a map of ChannelDescriptorShim objects, where each
+	// object wraps a reference to a legacy p2p ChannelDescriptor and the corresponding
+	// p2p proto.Message the new p2p Channel is responsible for handling.
+	//
+	//
+	// TODO: Remove once p2p refactor is complete.
+	// ref: https://github.com/tendermint/tendermint/issues/5670
+	ChannelShims = map[p2p.ChannelID]*p2p.ChannelDescriptorShim{
+		SnapshotChannel: {
+			MsgType: new(ssproto.Message),
+			Descriptor: &p2p.ChannelDescriptor{
+				ID:                  byte(SnapshotChannel),
+				Priority:            5,
+				SendQueueCapacity:   10,
+				RecvMessageCapacity: snapshotMsgSize,
+
+				MaxSendBytes: 400,
+			},
+		},
+		ChunkChannel: {
+			MsgType: new(ssproto.Message),
+			Descriptor: &p2p.ChannelDescriptor{
+				ID:                  byte(ChunkChannel),
+				Priority:            1,
+				SendQueueCapacity:   4,
+				RecvMessageCapacity: chunkMsgSize,
+
+				MaxSendBytes: 400,
+			},
+		},
+	}
+)
+
+const (
+	// SnapshotChannel exchanges snapshot metadata
+	SnapshotChannel = p2p.ChannelID(0x60)
+
+	// ChunkChannel exchanges chunk contents
+	ChunkChannel = p2p.ChannelID(0x61)
+
+	// recentSnapshots is the number of recent snapshots to send and receive per peer.
+	recentSnapshots = 10
+
+	// snapshotMsgSize is the maximum size of a snapshotResponseMessage
+	snapshotMsgSize = int(4e6)
+
+	// chunkMsgSize is the maximum size of a chunkResponseMessage
+	chunkMsgSize = int(16e6)
+)
+
+// Reactor handles state sync, both restoring snapshots for the local node and
+// serving snapshots for other nodes.
+type Reactor struct {
+	service.BaseService
+
+	conn        proxy.AppConnSnapshot
+	connQuery   proxy.AppConnQuery
+	tempDir     string
+	snapshotCh  *p2p.Channel
+	chunkCh     *p2p.Channel
+	peerUpdates *p2p.PeerUpdates
+	closeCh     chan struct{}
+
+	// This will only be set when a state sync is in progress. It is used to feed
+	// received snapshots and chunks into the sync.
+	mtx    tmsync.RWMutex
+	syncer *syncer
+}
+
+// NewReactor returns a reference to a new state sync reactor, which implements
+// the service.Service interface. It accepts a logger, connections for snapshots
+// and querying, references to p2p Channels and a channel to listen for peer
+// updates on. Note, the reactor will close all p2p Channels when stopping.
+func NewReactor(
+	logger log.Logger,
+	conn proxy.AppConnSnapshot,
+	connQuery proxy.AppConnQuery,
+	snapshotCh, chunkCh *p2p.Channel,
+	peerUpdates *p2p.PeerUpdates,
+	tempDir string,
+) *Reactor {
+	r := &Reactor{
+		conn:        conn,
+		connQuery:   connQuery,
+		snapshotCh:  snapshotCh,
+		chunkCh:     chunkCh,
+		peerUpdates: peerUpdates,
+		closeCh:     make(chan struct{}),
+		tempDir:     tempDir,
+	}
+
+	r.BaseService = *service.NewBaseService(logger, "StateSync", r)
+	return r
+}
+
+// OnStart starts separate go routines for each p2p Channel and listens for
+// envelopes on each. In addition, it also listens for peer updates and handles
+// messages on that p2p channel accordingly. The caller must be sure to execute
+// OnStop to ensure the outbound p2p Channels are closed. No error is returned.
+func (r *Reactor) OnStart() error {
+	// Listen for envelopes on the snapshot p2p Channel in a separate go-routine
+	// as to not block or cause IO contention with the chunk p2p Channel. Note,
+	// we do not launch a go-routine to handle individual envelopes as to not
+	// have to deal with bounding workers or pools.
+	go r.processSnapshotCh()
+
+	// Listen for envelopes on the chunk p2p Channel in a separate go-routine
+	// as to not block or cause IO contention with the snapshot p2p Channel. Note,
+	// we do not launch a go-routine to handle individual envelopes as to not
+	// have to deal with bounding workers or pools.
+	go r.processChunkCh()
+
+	go r.processPeerUpdates()
+
+	return nil
+}
+
+// OnStop stops the reactor by signaling to all spawned goroutines to exit and
+// blocking until they all exit.
+func (r *Reactor) OnStop() {
+	// Close closeCh to signal to all spawned goroutines to gracefully exit. All
+	// p2p Channels should execute Close().
+	close(r.closeCh)
+
+	// Wait for all p2p Channels to be closed before returning. This ensures we
+	// can easily reason about synchronization of all p2p Channels and ensure no
+	// panics will occur.
+	<-r.snapshotCh.Done()
+	<-r.chunkCh.Done()
+	<-r.peerUpdates.Done()
+}
+
+// handleSnapshotMessage handles envelopes sent from peers on the
+// SnapshotChannel. It returns an error only if the Envelope.Message is unknown
+// for this channel. This should never be called outside of handleMessage.
+func (r *Reactor) handleSnapshotMessage(envelope p2p.Envelope) error {
+	logger := r.Logger.With("peer", envelope.From)
+
+	switch msg := envelope.Message.(type) {
+	case *ssproto.SnapshotsRequest:
+		snapshots, err := r.recentSnapshots(recentSnapshots)
+		if err != nil {
+			logger.Error("failed to fetch snapshots", "err", err)
+			return nil
+		}
+
+		for _, snapshot := range snapshots {
+			logger.Debug(
+				"advertising snapshot",
+				"height", snapshot.Height,
+				"format", snapshot.Format,
+			)
+			r.snapshotCh.Out <- p2p.Envelope{
+				To: envelope.From,
+				Message: &ssproto.SnapshotsResponse{
+					Height:   snapshot.Height,
+					Format:   snapshot.Format,
+					Chunks:   snapshot.Chunks,
+					Hash:     snapshot.Hash,
+					Metadata: snapshot.Metadata,
+				},
+			}
+		}
+
+	case *ssproto.SnapshotsResponse:
+		r.mtx.RLock()
+		defer r.mtx.RUnlock()
+
+		if r.syncer == nil {
+			logger.Debug("received unexpected snapshot; no state sync in progress")
+			return nil
+		}
+
+		logger.Debug("received snapshot", "height", msg.Height, "format", msg.Format)
+		_, err := r.syncer.AddSnapshot(envelope.From, &snapshot{
+			Height:   msg.Height,
+			Format:   msg.Format,
+			Chunks:   msg.Chunks,
+			Hash:     msg.Hash,
+			Metadata: msg.Metadata,
+		})
+		if err != nil {
+			logger.Error(
+				"failed to add snapshot",
+				"height", msg.Height,
+				"format", msg.Format,
+				"err", err,
+				"channel", r.snapshotCh.ID,
+			)
+			return nil
+		}
+
+	default:
+		return fmt.Errorf("received unknown message: %T", msg)
+	}
+
+	return nil
+}
+
+// handleChunkMessage handles envelopes sent from peers on the ChunkChannel.
+// It returns an error only if the Envelope.Message is unknown for this channel.
+// This should never be called outside of handleMessage.
+func (r *Reactor) handleChunkMessage(envelope p2p.Envelope) error {
+	switch msg := envelope.Message.(type) {
+	case *ssproto.ChunkRequest:
+		r.Logger.Debug(
+			"received chunk request",
+			"height", msg.Height,
+			"format", msg.Format,
+			"chunk", msg.Index,
+			"peer", envelope.From,
+		)
+		resp, err := r.conn.LoadSnapshotChunkSync(context.Background(), abci.RequestLoadSnapshotChunk{
+			Height: msg.Height,
+			Format: msg.Format,
+			Chunk:  msg.Index,
+		})
+		if err != nil {
+			r.Logger.Error(
+				"failed to load chunk",
+				"height", msg.Height,
+				"format", msg.Format,
+				"chunk", msg.Index,
+				"err", err,
+				"peer", envelope.From,
+			)
+			return nil
+		}
+
+		r.Logger.Debug(
+			"sending chunk",
+			"height", msg.Height,
+			"format", msg.Format,
+			"chunk", msg.Index,
+			"peer", envelope.From,
+		)
+		r.chunkCh.Out <- p2p.Envelope{
+			To: envelope.From,
+			Message: &ssproto.ChunkResponse{
+				Height:  msg.Height,
+				Format:  msg.Format,
+				Index:   msg.Index,
+				Chunk:   resp.Chunk,
+				Missing: resp.Chunk == nil,
+			},
+		}
+
+	case *ssproto.ChunkResponse:
+		r.mtx.RLock()
+		defer r.mtx.RUnlock()
+
+		if r.syncer == nil {
+			r.Logger.Debug("received unexpected chunk; no state sync in progress", "peer", envelope.From)
+			return nil
+		}
+
+		r.Logger.Debug(
+			"received chunk; adding to sync",
+			"height", msg.Height,
+			"format", msg.Format,
+			"chunk", msg.Index,
+			"peer", envelope.From,
+		)
+		_, err := r.syncer.AddChunk(&chunk{
+			Height: msg.Height,
+			Format: msg.Format,
+			Index:  msg.Index,
+			Chunk:  msg.Chunk,
+			Sender: envelope.From,
+		})
+		if err != nil {
+			r.Logger.Error(
+				"failed to add chunk",
+				"height", msg.Height,
+				"format", msg.Format,
+				"chunk", msg.Index,
+				"err", err,
+				"peer", envelope.From,
+			)
+			return nil
+		}
+
+	default:
+		return fmt.Errorf("received unknown message: %T", msg)
+	}
+
+	return nil
+}
+
+// handleMessage handles an Envelope sent from a peer on a specific p2p Channel.
+// It will handle errors and any possible panics gracefully. A caller can handle
+// any error returned by sending a PeerError on the respective channel.
+func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err error) {
+	defer func() {
+		if e := recover(); e != nil {
+			err = fmt.Errorf("panic in processing message: %v", e)
+		}
+	}()
+
+	r.Logger.Debug("received message", "message", envelope.Message, "peer", envelope.From)
+
+	switch chID {
+	case SnapshotChannel:
+		err = r.handleSnapshotMessage(envelope)
+
+	case ChunkChannel:
+		err = r.handleChunkMessage(envelope)
+
+	default:
+		err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
+	}
+
+	return err
+}
+
+// processSnapshotCh initiates a blocking process where we listen for and handle
+// envelopes on the SnapshotChannel. Any error encountered during message
+// execution will result in a PeerError being sent on the SnapshotChannel. When
+// the reactor is stopped, we will catch the signal and close the p2p Channel
+// gracefully.
+func (r *Reactor) processSnapshotCh() {
+	defer r.snapshotCh.Close()
+
+	for {
+		select {
+		case envelope := <-r.snapshotCh.In:
+			if err := r.handleMessage(r.snapshotCh.ID, envelope); err != nil {
+				r.Logger.Error("failed to process message", "ch_id", r.snapshotCh.ID, "envelope", envelope, "err", err)
+				r.snapshotCh.Error <- p2p.PeerError{
+					NodeID: envelope.From,
+					Err:    err,
+				}
+			}
+
+		case <-r.closeCh:
+			r.Logger.Debug("stopped listening on snapshot channel; closing...")
+			return
+		}
+	}
+}
+
+// processChunkCh initiates a blocking process where we listen for and handle
+// envelopes on the ChunkChannel. Any error encountered during message
+// execution will result in a PeerError being sent on the ChunkChannel. When
+// the reactor is stopped, we will catch the signal and close the p2p Channel
+// gracefully.
+func (r *Reactor) processChunkCh() {
+	defer r.chunkCh.Close()
+
+	for {
+		select {
+		case envelope := <-r.chunkCh.In:
+			if err := r.handleMessage(r.chunkCh.ID, envelope); err != nil {
+				r.Logger.Error("failed to process message", "ch_id", r.chunkCh.ID, "envelope", envelope, "err", err)
+				r.chunkCh.Error <- p2p.PeerError{
+					NodeID: envelope.From,
+					Err:    err,
+				}
+			}
+
+		case <-r.closeCh:
+			r.Logger.Debug("stopped listening on chunk channel; closing...")
+			return
+		}
+	}
+}
+
+// processPeerUpdate processes a PeerUpdate, returning an error upon failing to
+// handle the PeerUpdate or if a panic is recovered.
+func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
+	r.Logger.Debug("received peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
+
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
+
+	if r.syncer != nil {
+		switch peerUpdate.Status {
+		case p2p.PeerStatusUp:
+			r.syncer.AddPeer(peerUpdate.NodeID)
+
+		case p2p.PeerStatusDown:
+			r.syncer.RemovePeer(peerUpdate.NodeID)
+		}
+	}
+}
+
+// processPeerUpdates initiates a blocking process where we listen for and handle
+// PeerUpdate messages. When the reactor is stopped, we will catch the signal and
+// close the p2p PeerUpdatesCh gracefully.
+func (r *Reactor) processPeerUpdates() {
+	defer r.peerUpdates.Close()
+
+	for {
+		select {
+		case peerUpdate := <-r.peerUpdates.Updates():
+			r.processPeerUpdate(peerUpdate)
+
+		case <-r.closeCh:
+			r.Logger.Debug("stopped listening on peer updates channel; closing...")
+			return
+		}
+	}
+}
+
+// recentSnapshots fetches the n most recent snapshots from the app
+func (r *Reactor) recentSnapshots(n uint32) ([]*snapshot, error) {
+	resp, err := r.conn.ListSnapshotsSync(context.Background(), abci.RequestListSnapshots{})
+	if err != nil {
+		return nil, err
+	}
+
+	sort.Slice(resp.Snapshots, func(i, j int) bool {
+		a := resp.Snapshots[i]
+		b := resp.Snapshots[j]
+
+		switch {
+		case a.Height > b.Height:
+			return true
+		case a.Height == b.Height && a.Format > b.Format:
+			return true
+		default:
+			return false
+		}
+	})
+
+	snapshots := make([]*snapshot, 0, n)
+	for i, s := range resp.Snapshots {
+		if i >= recentSnapshots {
+			break
+		}
+
+		snapshots = append(snapshots, &snapshot{
+			Height:   s.Height,
+			Format:   s.Format,
+			Chunks:   s.Chunks,
+			Hash:     s.Hash,
+			Metadata: s.Metadata,
+		})
+	}
+
+	return snapshots, nil
+}
+
+// Sync runs a state sync, returning the new state and last commit at the snapshot height.
+// The caller must store the state and commit in the state database and block store.
+func (r *Reactor) Sync(stateProvider StateProvider, discoveryTime time.Duration) (sm.State, *types.Commit, error) {
+	r.mtx.Lock()
+	if r.syncer != nil {
+		r.mtx.Unlock()
+		return sm.State{}, nil, errors.New("a state sync is already in progress")
+	}
+
+	r.syncer = newSyncer(r.Logger, r.conn, r.connQuery, stateProvider, r.snapshotCh.Out, r.chunkCh.Out, r.tempDir)
+	r.mtx.Unlock()
+
+	hook := func() {
+		// request snapshots from all currently connected peers
+		r.Logger.Debug("requesting snapshots from known peers")
+		r.snapshotCh.Out <- p2p.Envelope{
+			Broadcast: true,
+			Message:   &ssproto.SnapshotsRequest{},
+		}
+	}
+
+	hook()
+
+	state, commit, err := r.syncer.SyncAny(discoveryTime, hook)
+
+	r.mtx.Lock()
+	r.syncer = nil
+	r.mtx.Unlock()
+
+	return state, commit, err
+}

internal/statesync/reactor_test.go

@@ -0,0 +1,286 @@
+package statesync
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/statesync/mocks"
+	"github.com/tendermint/tendermint/libs/log"
+	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
+	proxymocks "github.com/tendermint/tendermint/proxy/mocks"
+)
+
+type reactorTestSuite struct {
+	reactor *Reactor
+	syncer  *syncer
+
+	conn          *proxymocks.AppConnSnapshot
+	connQuery     *proxymocks.AppConnQuery
+	stateProvider *mocks.StateProvider
+
+	snapshotChannel   *p2p.Channel
+	snapshotInCh      chan p2p.Envelope
+	snapshotOutCh     chan p2p.Envelope
+	snapshotPeerErrCh chan p2p.PeerError
+
+	chunkChannel   *p2p.Channel
+	chunkInCh      chan p2p.Envelope
+	chunkOutCh     chan p2p.Envelope
+	chunkPeerErrCh chan p2p.PeerError
+
+	peerUpdates *p2p.PeerUpdates
+}
+
+func setup(
+	t *testing.T,
+	conn *proxymocks.AppConnSnapshot,
+	connQuery *proxymocks.AppConnQuery,
+	stateProvider *mocks.StateProvider,
+	chBuf uint,
+) *reactorTestSuite {
+	t.Helper()
+
+	if conn == nil {
+		conn = &proxymocks.AppConnSnapshot{}
+	}
+	if connQuery == nil {
+		connQuery = &proxymocks.AppConnQuery{}
+	}
+	if stateProvider == nil {
+		stateProvider = &mocks.StateProvider{}
+	}
+
+	rts := &reactorTestSuite{
+		snapshotInCh:      make(chan p2p.Envelope, chBuf),
+		snapshotOutCh:     make(chan p2p.Envelope, chBuf),
+		snapshotPeerErrCh: make(chan p2p.PeerError, chBuf),
+		chunkInCh:         make(chan p2p.Envelope, chBuf),
+		chunkOutCh:        make(chan p2p.Envelope, chBuf),
+		chunkPeerErrCh:    make(chan p2p.PeerError, chBuf),
+		peerUpdates:       p2p.NewPeerUpdates(make(chan p2p.PeerUpdate), int(chBuf)),
+		conn:              conn,
+		connQuery:         connQuery,
+		stateProvider:     stateProvider,
+	}
+
+	rts.snapshotChannel = p2p.NewChannel(
+		SnapshotChannel,
+		new(ssproto.Message),
+		rts.snapshotInCh,
+		rts.snapshotOutCh,
+		rts.snapshotPeerErrCh,
+	)
+
+	rts.chunkChannel = p2p.NewChannel(
+		ChunkChannel,
+		new(ssproto.Message),
+		rts.chunkInCh,
+		rts.chunkOutCh,
+		rts.chunkPeerErrCh,
+	)
+
+	rts.reactor = NewReactor(
+		log.NewNopLogger(),
+		conn,
+		connQuery,
+		rts.snapshotChannel,
+		rts.chunkChannel,
+		rts.peerUpdates,
+		"",
+	)
+
+	rts.syncer = newSyncer(
+		log.NewNopLogger(),
+		conn,
+		connQuery,
+		stateProvider,
+		rts.snapshotOutCh,
+		rts.chunkOutCh,
+		"",
+	)
+
+	require.NoError(t, rts.reactor.Start())
+	require.True(t, rts.reactor.IsRunning())
+
+	t.Cleanup(func() {
+		require.NoError(t, rts.reactor.Stop())
+		require.False(t, rts.reactor.IsRunning())
+	})
+
+	return rts
+}
+
+func TestReactor_ChunkRequest_InvalidRequest(t *testing.T) {
+	rts := setup(t, nil, nil, nil, 2)
+
+	rts.chunkInCh <- p2p.Envelope{
+		From:    p2p.NodeID("aa"),
+		Message: &ssproto.SnapshotsRequest{},
+	}
+
+	response := <-rts.chunkPeerErrCh
+	require.Error(t, response.Err)
+	require.Empty(t, rts.chunkOutCh)
+	require.Contains(t, response.Err.Error(), "received unknown message")
+	require.Equal(t, p2p.NodeID("aa"), response.NodeID)
+}
+
+func TestReactor_ChunkRequest(t *testing.T) {
+	testcases := map[string]struct {
+		request        *ssproto.ChunkRequest
+		chunk          []byte
+		expectResponse *ssproto.ChunkResponse
+	}{
+		"chunk is returned": {
+			&ssproto.ChunkRequest{Height: 1, Format: 1, Index: 1},
+			[]byte{1, 2, 3},
+			&ssproto.ChunkResponse{Height: 1, Format: 1, Index: 1, Chunk: []byte{1, 2, 3}},
+		},
+		"empty chunk is returned, as empty": {
+			&ssproto.ChunkRequest{Height: 1, Format: 1, Index: 1},
+			[]byte{},
+			&ssproto.ChunkResponse{Height: 1, Format: 1, Index: 1, Chunk: []byte{}},
+		},
+		"nil (missing) chunk is returned as missing": {
+			&ssproto.ChunkRequest{Height: 1, Format: 1, Index: 1},
+			nil,
+			&ssproto.ChunkResponse{Height: 1, Format: 1, Index: 1, Missing: true},
+		},
+		"invalid request": {
+			&ssproto.ChunkRequest{Height: 1, Format: 1, Index: 1},
+			nil,
+			&ssproto.ChunkResponse{Height: 1, Format: 1, Index: 1, Missing: true},
+		},
+	}
+
+	for name, tc := range testcases {
+		tc := tc
+
+		t.Run(name, func(t *testing.T) {
+			// mock ABCI connection to return local snapshots
+			conn := &proxymocks.AppConnSnapshot{}
+			conn.On("LoadSnapshotChunkSync", context.Background(), abci.RequestLoadSnapshotChunk{
+				Height: tc.request.Height,
+				Format: tc.request.Format,
+				Chunk:  tc.request.Index,
+			}).Return(&abci.ResponseLoadSnapshotChunk{Chunk: tc.chunk}, nil)
+
+			rts := setup(t, conn, nil, nil, 2)
+
+			rts.chunkInCh <- p2p.Envelope{
+				From:    p2p.NodeID("aa"),
+				Message: tc.request,
+			}
+
+			response := <-rts.chunkOutCh
+			require.Equal(t, tc.expectResponse, response.Message)
+			require.Empty(t, rts.chunkOutCh)
+
+			conn.AssertExpectations(t)
+		})
+	}
+}
+
+func TestReactor_SnapshotsRequest_InvalidRequest(t *testing.T) {
+	rts := setup(t, nil, nil, nil, 2)
+
+	rts.snapshotInCh <- p2p.Envelope{
+		From:    p2p.NodeID("aa"),
+		Message: &ssproto.ChunkRequest{},
+	}
+
+	response := <-rts.snapshotPeerErrCh
+	require.Error(t, response.Err)
+	require.Empty(t, rts.snapshotOutCh)
+	require.Contains(t, response.Err.Error(), "received unknown message")
+	require.Equal(t, p2p.NodeID("aa"), response.NodeID)
+}
+
+func TestReactor_SnapshotsRequest(t *testing.T) {
+	testcases := map[string]struct {
+		snapshots       []*abci.Snapshot
+		expectResponses []*ssproto.SnapshotsResponse
+	}{
+		"no snapshots": {nil, []*ssproto.SnapshotsResponse{}},
+		">10 unordered snapshots": {
+			[]*abci.Snapshot{
+				{Height: 1, Format: 2, Chunks: 7, Hash: []byte{1, 2}, Metadata: []byte{1}},
+				{Height: 2, Format: 2, Chunks: 7, Hash: []byte{2, 2}, Metadata: []byte{2}},
+				{Height: 3, Format: 2, Chunks: 7, Hash: []byte{3, 2}, Metadata: []byte{3}},
+				{Height: 1, Format: 1, Chunks: 7, Hash: []byte{1, 1}, Metadata: []byte{4}},
+				{Height: 2, Format: 1, Chunks: 7, Hash: []byte{2, 1}, Metadata: []byte{5}},
+				{Height: 3, Format: 1, Chunks: 7, Hash: []byte{3, 1}, Metadata: []byte{6}},
+				{Height: 1, Format: 4, Chunks: 7, Hash: []byte{1, 4}, Metadata: []byte{7}},
+				{Height: 2, Format: 4, Chunks: 7, Hash: []byte{2, 4}, Metadata: []byte{8}},
+				{Height: 3, Format: 4, Chunks: 7, Hash: []byte{3, 4}, Metadata: []byte{9}},
+				{Height: 1, Format: 3, Chunks: 7, Hash: []byte{1, 3}, Metadata: []byte{10}},
+				{Height: 2, Format: 3, Chunks: 7, Hash: []byte{2, 3}, Metadata: []byte{11}},
+				{Height: 3, Format: 3, Chunks: 7, Hash: []byte{3, 3}, Metadata: []byte{12}},
+			},
+			[]*ssproto.SnapshotsResponse{
+				{Height: 3, Format: 4, Chunks: 7, Hash: []byte{3, 4}, Metadata: []byte{9}},
+				{Height: 3, Format: 3, Chunks: 7, Hash: []byte{3, 3}, Metadata: []byte{12}},
+				{Height: 3, Format: 2, Chunks: 7, Hash: []byte{3, 2}, Metadata: []byte{3}},
+				{Height: 3, Format: 1, Chunks: 7, Hash: []byte{3, 1}, Metadata: []byte{6}},
+				{Height: 2, Format: 4, Chunks: 7, Hash: []byte{2, 4}, Metadata: []byte{8}},
+				{Height: 2, Format: 3, Chunks: 7, Hash: []byte{2, 3}, Metadata: []byte{11}},
+				{Height: 2, Format: 2, Chunks: 7, Hash: []byte{2, 2}, Metadata: []byte{2}},
+				{Height: 2, Format: 1, Chunks: 7, Hash: []byte{2, 1}, Metadata: []byte{5}},
+				{Height: 1, Format: 4, Chunks: 7, Hash: []byte{1, 4}, Metadata: []byte{7}},
+				{Height: 1, Format: 3, Chunks: 7, Hash: []byte{1, 3}, Metadata: []byte{10}},
+			},
+		},
+	}
+
+	for name, tc := range testcases {
+		tc := tc
+
+		t.Run(name, func(t *testing.T) {
+			// mock ABCI connection to return local snapshots
+			conn := &proxymocks.AppConnSnapshot{}
+			conn.On("ListSnapshotsSync", context.Background(), abci.RequestListSnapshots{}).Return(&abci.ResponseListSnapshots{
+				Snapshots: tc.snapshots,
+			}, nil)
+
+			rts := setup(t, conn, nil, nil, 100)
+
+			rts.snapshotInCh <- p2p.Envelope{
+				From:    p2p.NodeID("aa"),
+				Message: &ssproto.SnapshotsRequest{},
+			}
+
+			if len(tc.expectResponses) > 0 {
+				retryUntil(t, func() bool { return len(rts.snapshotOutCh) == len(tc.expectResponses) }, time.Second)
+			}
+
+			responses := make([]*ssproto.SnapshotsResponse, len(tc.expectResponses))
+			for i := 0; i < len(tc.expectResponses); i++ {
+				e := <-rts.snapshotOutCh
+				responses[i] = e.Message.(*ssproto.SnapshotsResponse)
+			}
+
+			require.Equal(t, tc.expectResponses, responses)
+			require.Empty(t, rts.snapshotOutCh)
+		})
+	}
+}
+
+// retryUntil will continue to evaluate fn and will return successfully when true
+// or fail when the timeout is reached.
+func retryUntil(t *testing.T, fn func() bool, timeout time.Duration) {
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	for {
+		if fn() {
+			return
+		}
+
+		require.NoError(t, ctx.Err())
+	}
+}

internal/statesync/snapshots.go

@@ -0,0 +1,302 @@
+package statesync
+
+import (
+	"context"
+	"crypto/sha256"
+	"fmt"
+	"math/rand"
+	"sort"
+	"strings"
+	"time"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+	"github.com/tendermint/tendermint/internal/p2p"
+)
+
+// snapshotKey is a snapshot key used for lookups.
+type snapshotKey [sha256.Size]byte
+
+// snapshot contains data about a snapshot.
+type snapshot struct {
+	Height   uint64
+	Format   uint32
+	Chunks   uint32
+	Hash     []byte
+	Metadata []byte
+
+	trustedAppHash []byte // populated by light client
+}
+
+// Key generates a snapshot key, used for lookups. It takes into account not only the height and
+// format, but also the chunks, hash, and metadata in case peers have generated snapshots in a
+// non-deterministic manner. All fields must be equal for the snapshot to be considered the same.
+func (s *snapshot) Key() snapshotKey {
+	// Hash.Write() never returns an error.
+	hasher := sha256.New()
+	hasher.Write([]byte(fmt.Sprintf("%v:%v:%v", s.Height, s.Format, s.Chunks)))
+	hasher.Write(s.Hash)
+	hasher.Write(s.Metadata)
+	var key snapshotKey
+	copy(key[:], hasher.Sum(nil))
+	return key
+}
+
+// snapshotPool discovers and aggregates snapshots across peers.
+type snapshotPool struct {
+	stateProvider StateProvider
+
+	tmsync.Mutex
+	snapshots     map[snapshotKey]*snapshot
+	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[p2p.NodeID]map[snapshotKey]bool
+
+	// blacklists for rejected items
+	formatBlacklist   map[uint32]bool
+	peerBlacklist     map[p2p.NodeID]bool
+	snapshotBlacklist map[snapshotKey]bool
+}
+
+// newSnapshotPool creates a new snapshot pool. The state source is used for
+func newSnapshotPool(stateProvider StateProvider) *snapshotPool {
+	return &snapshotPool{
+		stateProvider:     stateProvider,
+		snapshots:         make(map[snapshotKey]*snapshot),
+		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[p2p.NodeID]map[snapshotKey]bool),
+		formatBlacklist:   make(map[uint32]bool),
+		peerBlacklist:     make(map[p2p.NodeID]bool),
+		snapshotBlacklist: make(map[snapshotKey]bool),
+	}
+}
+
+// Add adds a snapshot to the pool, unless the peer has already sent recentSnapshots
+// 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 p2p.NodeID, snapshot *snapshot) (bool, error) {
+	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
+	}
+	snapshot.trustedAppHash = appHash
+	key := snapshot.Key()
+
+	p.Lock()
+	defer p.Unlock()
+
+	switch {
+	case p.formatBlacklist[snapshot.Format]:
+		return false, nil
+	case p.peerBlacklist[peerID]:
+		return false, nil
+	case p.snapshotBlacklist[key]:
+		return false, nil
+	case len(p.peerIndex[peerID]) >= recentSnapshots:
+		return false, nil
+	}
+
+	if p.snapshotPeers[key] == nil {
+		p.snapshotPeers[key] = make(map[p2p.NodeID]p2p.NodeID)
+	}
+	p.snapshotPeers[key][peerID] = peerID
+
+	if p.peerIndex[peerID] == nil {
+		p.peerIndex[peerID] = make(map[snapshotKey]bool)
+	}
+	p.peerIndex[peerID][key] = true
+
+	if p.snapshots[key] != nil {
+		return false, nil
+	}
+	p.snapshots[key] = snapshot
+
+	if p.formatIndex[snapshot.Format] == nil {
+		p.formatIndex[snapshot.Format] = make(map[snapshotKey]bool)
+	}
+	p.formatIndex[snapshot.Format][key] = true
+
+	if p.heightIndex[snapshot.Height] == nil {
+		p.heightIndex[snapshot.Height] = make(map[snapshotKey]bool)
+	}
+	p.heightIndex[snapshot.Height][key] = true
+
+	return true, nil
+}
+
+// Best returns the "best" currently known snapshot, if any.
+func (p *snapshotPool) Best() *snapshot {
+	ranked := p.Ranked()
+	if len(ranked) == 0 {
+		return nil
+	}
+	return ranked[0]
+}
+
+// GetPeer returns a random peer for a snapshot, if any.
+func (p *snapshotPool) GetPeer(snapshot *snapshot) p2p.NodeID {
+	peers := p.GetPeers(snapshot)
+	if len(peers) == 0 {
+		return ""
+	}
+	return peers[rand.Intn(len(peers))] // nolint:gosec // G404: Use of weak random number generator
+}
+
+// GetPeers returns the peers for a snapshot.
+func (p *snapshotPool) GetPeers(snapshot *snapshot) []p2p.NodeID {
+	key := snapshot.Key()
+
+	p.Lock()
+	defer p.Unlock()
+
+	peers := make([]p2p.NodeID, 0, len(p.snapshotPeers[key]))
+	for _, peer := range p.snapshotPeers[key] {
+		peers = append(peers, peer)
+	}
+
+	// sort results, for testability (otherwise order is random, so tests randomly fail)
+	sort.Slice(peers, func(a int, b int) bool {
+		return strings.Compare(string(peers[a]), string(peers[b])) < 0
+	})
+
+	return peers
+}
+
+// Ranked returns a list of snapshots ranked by preference. The current heuristic is very naïve,
+// preferring the snapshot with the greatest height, then greatest format, then greatest number of
+// peers. This can be improved quite a lot.
+func (p *snapshotPool) Ranked() []*snapshot {
+	p.Lock()
+	defer p.Unlock()
+
+	if len(p.snapshots) == 0 {
+		return []*snapshot{}
+	}
+
+	numPeers := make([]int, 0, len(p.snapshots))
+	for key := range p.snapshots {
+		numPeers = append(numPeers, len(p.snapshotPeers[key]))
+	}
+	sort.Ints(numPeers)
+	median := len(numPeers) / 2
+	if len(numPeers)%2 == 0 {
+		median = (numPeers[median-1] + numPeers[median]) / 2
+	} else {
+		median = numPeers[median]
+	}
+
+	commonCandidates := make([]*snapshot, 0, len(p.snapshots)/2)
+	uncommonCandidates := make([]*snapshot, 0, len(p.snapshots)/2)
+	for key := range p.snapshots {
+		if len(p.snapshotPeers[key]) > median {
+			commonCandidates = append(commonCandidates, p.snapshots[key])
+			continue
+		}
+
+		uncommonCandidates = append(uncommonCandidates, p.snapshots[key])
+	}
+
+	sort.Slice(commonCandidates, p.sorterFactory(commonCandidates))
+	sort.Slice(uncommonCandidates, p.sorterFactory(uncommonCandidates))
+
+	return append(commonCandidates, uncommonCandidates...)
+}
+
+func (p *snapshotPool) sorterFactory(candidates []*snapshot) func(int, int) bool {
+	return func(i, j int) bool {
+		a := candidates[i]
+		b := candidates[j]
+
+		switch {
+		case a.Height > b.Height:
+			return true
+		case a.Height < b.Height:
+			return false
+		case len(p.snapshotPeers[a.Key()]) > len(p.snapshotPeers[b.Key()]):
+			return true
+		case a.Format > b.Format:
+			return true
+		case a.Format < b.Format:
+			return false
+		default:
+			return false
+		}
+	}
+}
+
+// Reject rejects a snapshot. Rejected snapshots will never be used again.
+func (p *snapshotPool) Reject(snapshot *snapshot) {
+	key := snapshot.Key()
+	p.Lock()
+	defer p.Unlock()
+
+	p.snapshotBlacklist[key] = true
+	p.removeSnapshot(key)
+}
+
+// RejectFormat rejects a snapshot format. It will never be used again.
+func (p *snapshotPool) RejectFormat(format uint32) {
+	p.Lock()
+	defer p.Unlock()
+
+	p.formatBlacklist[format] = true
+	for key := range p.formatIndex[format] {
+		p.removeSnapshot(key)
+	}
+}
+
+// RejectPeer rejects a peer. It will never be used again.
+func (p *snapshotPool) RejectPeer(peerID p2p.NodeID) {
+	if len(peerID) == 0 {
+		return
+	}
+
+	p.Lock()
+	defer p.Unlock()
+
+	p.removePeer(peerID)
+	p.peerBlacklist[peerID] = true
+}
+
+// RemovePeer removes a peer from the pool, and any snapshots that no longer have peers.
+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 p2p.NodeID) {
+	for key := range p.peerIndex[peerID] {
+		delete(p.snapshotPeers[key], peerID)
+		if len(p.snapshotPeers[key]) == 0 {
+			p.removeSnapshot(key)
+		}
+	}
+
+	delete(p.peerIndex, peerID)
+}
+
+// removeSnapshot removes a snapshot. The caller must hold the mutex lock.
+func (p *snapshotPool) removeSnapshot(key snapshotKey) {
+	snapshot := p.snapshots[key]
+	if snapshot == nil {
+		return
+	}
+
+	delete(p.snapshots, key)
+	delete(p.formatIndex[snapshot.Format], key)
+	delete(p.heightIndex[snapshot.Height], key)
+	for peerID := range p.snapshotPeers[key] {
+		delete(p.peerIndex[peerID], key)
+	}
+	delete(p.snapshotPeers, key)
+}

internal/statesync/snapshots_test.go

@@ -0,0 +1,324 @@
+package statesync
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/internal/statesync/mocks"
+)
+
+func TestSnapshot_Key(t *testing.T) {
+	testcases := map[string]struct {
+		modify func(*snapshot)
+	}{
+		"new height":      {func(s *snapshot) { s.Height = 9 }},
+		"new format":      {func(s *snapshot) { s.Format = 9 }},
+		"new chunk count": {func(s *snapshot) { s.Chunks = 9 }},
+		"new hash":        {func(s *snapshot) { s.Hash = []byte{9} }},
+		"no metadata":     {func(s *snapshot) { s.Metadata = nil }},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			s := snapshot{
+				Height:   3,
+				Format:   1,
+				Chunks:   7,
+				Hash:     []byte{1, 2, 3},
+				Metadata: []byte{255},
+			}
+			before := s.Key()
+			tc.modify(&s)
+			after := s.Key()
+			require.NotEqual(t, before, after)
+		})
+	}
+}
+
+func TestSnapshotPool_Add(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, uint64(1)).Return([]byte("app_hash"), nil)
+
+	peerID := p2p.NodeID("aa")
+
+	// Adding to the pool should work
+	pool := newSnapshotPool(stateProvider)
+	added, err := pool.Add(peerID, &snapshot{
+		Height: 1,
+		Format: 1,
+		Chunks: 1,
+		Hash:   []byte{1},
+	})
+	require.NoError(t, err)
+	require.True(t, added)
+
+	// Adding again from a different peer should return false
+	otherNodeID := p2p.NodeID("bb")
+	added, err = pool.Add(otherNodeID, &snapshot{
+		Height: 1,
+		Format: 1,
+		Chunks: 1,
+		Hash:   []byte{1},
+	})
+	require.NoError(t, err)
+	require.False(t, added)
+
+	// The pool should have populated the snapshot with the trusted app hash
+	snapshot := pool.Best()
+	require.NotNil(t, snapshot)
+	require.Equal(t, []byte("app_hash"), snapshot.trustedAppHash)
+
+	stateProvider.AssertExpectations(t)
+}
+
+func TestSnapshotPool_GetPeer(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	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}}
+
+	peerAID := p2p.NodeID("aa")
+	peerBID := p2p.NodeID("bb")
+
+	_, err := pool.Add(peerAID, s)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerBID, s)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerAID, &snapshot{Height: 2, Format: 1, Chunks: 1, Hash: []byte{1}})
+	require.NoError(t, err)
+
+	// GetPeer currently picks a random peer, so lets run it until we've seen both.
+	seenA := false
+	seenB := false
+	for !seenA || !seenB {
+		peer := pool.GetPeer(s)
+		if peer == peerAID {
+			seenA = true
+		}
+		if peer == peerBID {
+			seenB = true
+		}
+	}
+
+	// GetPeer should return empty for an unknown snapshot
+	peer := pool.GetPeer(&snapshot{Height: 9, Format: 9})
+	require.EqualValues(t, "", peer)
+}
+
+func TestSnapshotPool_GetPeers(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	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}}
+
+	peerAID := p2p.NodeID("aa")
+	peerBID := p2p.NodeID("bb")
+
+	_, err := pool.Add(peerAID, s)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerBID, s)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerAID, &snapshot{Height: 2, Format: 1, Chunks: 1, Hash: []byte{2}})
+	require.NoError(t, err)
+
+	peers := pool.GetPeers(s)
+	require.Len(t, peers, 2)
+	require.Equal(t, peerAID, peers[0])
+	require.EqualValues(t, peerBID, peers[1])
+}
+
+func TestSnapshotPool_Ranked_Best(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	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.
+	// Snapshots with different chunk hashes are considered different, and the most peers is
+	// tie-breaker.
+	expectSnapshots := []struct {
+		snapshot *snapshot
+		peers    []p2p.NodeID
+	}{
+		{&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.
+	for i := len(expectSnapshots) - 1; i >= 0; i-- {
+		for _, peerID := range expectSnapshots[i].peers {
+			_, err := pool.Add(peerID, expectSnapshots[i].snapshot)
+			require.NoError(t, err)
+		}
+	}
+
+	// Ranked should return the snapshots in the same order
+	ranked := pool.Ranked()
+	require.Len(t, ranked, len(expectSnapshots))
+
+	for i := range ranked {
+		require.Equal(t, expectSnapshots[i].snapshot, ranked[i])
+	}
+
+	// Check that best snapshots are returned in expected order
+	for i := range expectSnapshots {
+		snapshot := expectSnapshots[i].snapshot
+		require.Equal(t, snapshot, pool.Best())
+		pool.Reject(snapshot)
+	}
+
+	require.Nil(t, pool.Best())
+}
+
+func TestSnapshotPool_Reject(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+	pool := newSnapshotPool(stateProvider)
+
+	peerID := p2p.NodeID("aa")
+
+	snapshots := []*snapshot{
+		{Height: 2, Format: 2, Chunks: 1, Hash: []byte{1, 2}},
+		{Height: 2, Format: 1, Chunks: 1, Hash: []byte{1, 2}},
+		{Height: 1, Format: 2, Chunks: 1, Hash: []byte{1, 2}},
+		{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1, 2}},
+	}
+	for _, s := range snapshots {
+		_, err := pool.Add(peerID, s)
+		require.NoError(t, err)
+	}
+
+	pool.Reject(snapshots[0])
+	require.Equal(t, snapshots[1:], pool.Ranked())
+
+	added, err := pool.Add(peerID, snapshots[0])
+	require.NoError(t, err)
+	require.False(t, added)
+
+	added, err = pool.Add(peerID, &snapshot{Height: 3, Format: 3, Chunks: 1, Hash: []byte{1}})
+	require.NoError(t, err)
+	require.True(t, added)
+}
+
+func TestSnapshotPool_RejectFormat(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+	pool := newSnapshotPool(stateProvider)
+
+	peerID := p2p.NodeID("aa")
+
+	snapshots := []*snapshot{
+		{Height: 2, Format: 2, Chunks: 1, Hash: []byte{1, 2}},
+		{Height: 2, Format: 1, Chunks: 1, Hash: []byte{1, 2}},
+		{Height: 1, Format: 2, Chunks: 1, Hash: []byte{1, 2}},
+		{Height: 1, Format: 1, Chunks: 1, Hash: []byte{1, 2}},
+	}
+	for _, s := range snapshots {
+		_, err := pool.Add(peerID, s)
+		require.NoError(t, err)
+	}
+
+	pool.RejectFormat(1)
+	require.Equal(t, []*snapshot{snapshots[0], snapshots[2]}, pool.Ranked())
+
+	added, err := pool.Add(peerID, &snapshot{Height: 3, Format: 1, Chunks: 1, Hash: []byte{1}})
+	require.NoError(t, err)
+	require.False(t, added)
+	require.Equal(t, []*snapshot{snapshots[0], snapshots[2]}, pool.Ranked())
+
+	added, err = pool.Add(peerID, &snapshot{Height: 3, Format: 3, Chunks: 1, Hash: []byte{1}})
+	require.NoError(t, err)
+	require.True(t, added)
+}
+
+func TestSnapshotPool_RejectPeer(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+	pool := newSnapshotPool(stateProvider)
+
+	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}}
+	s3 := &snapshot{Height: 3, Format: 1, Chunks: 1, Hash: []byte{2}}
+
+	_, err := pool.Add(peerAID, s1)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerAID, s2)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerBID, s2)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerBID, s3)
+	require.NoError(t, err)
+
+	pool.RejectPeer(peerAID)
+
+	require.Empty(t, pool.GetPeers(s1))
+
+	peers2 := pool.GetPeers(s2)
+	require.Len(t, peers2, 1)
+	require.Equal(t, peerBID, peers2[0])
+
+	peers3 := pool.GetPeers(s2)
+	require.Len(t, peers3, 1)
+	require.Equal(t, peerBID, peers3[0])
+
+	// it should no longer be possible to add the peer back
+	_, err = pool.Add(peerAID, s1)
+	require.NoError(t, err)
+	require.Empty(t, pool.GetPeers(s1))
+}
+
+func TestSnapshotPool_RemovePeer(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+	pool := newSnapshotPool(stateProvider)
+
+	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}}
+
+	_, err := pool.Add(peerAID, s1)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerAID, s2)
+	require.NoError(t, err)
+
+	_, err = pool.Add(peerBID, s1)
+	require.NoError(t, err)
+
+	pool.RemovePeer(peerAID)
+
+	peers1 := pool.GetPeers(s1)
+	require.Len(t, peers1, 1)
+	require.Equal(t, peerBID, peers1[0])
+
+	peers2 := pool.GetPeers(s2)
+	require.Empty(t, peers2)
+
+	// it should still be possible to add the peer back
+	_, err = pool.Add(peerAID, s1)
+	require.NoError(t, err)
+
+	peers1 = pool.GetPeers(s1)
+	require.Len(t, peers1, 2)
+	require.Equal(t, peerAID, peers1[0])
+	require.Equal(t, peerBID, peers1[1])
+}

internal/statesync/stateprovider.go

@@ -0,0 +1,202 @@
+package statesync
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	dbm "github.com/tendermint/tm-db"
+
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+	"github.com/tendermint/tendermint/libs/log"
+	"github.com/tendermint/tendermint/light"
+	lightprovider "github.com/tendermint/tendermint/light/provider"
+	lighthttp "github.com/tendermint/tendermint/light/provider/http"
+	lightrpc "github.com/tendermint/tendermint/light/rpc"
+	lightdb "github.com/tendermint/tendermint/light/store/db"
+	rpchttp "github.com/tendermint/tendermint/rpc/client/http"
+	sm "github.com/tendermint/tendermint/state"
+	"github.com/tendermint/tendermint/types"
+)
+
+//go:generate mockery --case underscore --name StateProvider
+
+// StateProvider is a provider of trusted state data for bootstrapping a node. This refers
+// 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(ctx context.Context, height uint64) ([]byte, error)
+	// Commit returns the commit at the given height.
+	Commit(ctx context.Context, height uint64) (*types.Commit, error)
+	// State returns a state object at the given height.
+	State(ctx context.Context, height uint64) (sm.State, error)
+}
+
+// lightClientStateProvider is a state provider using the light client.
+type lightClientStateProvider struct {
+	tmsync.Mutex  // light.Client is not concurrency-safe
+	lc            *light.Client
+	version       sm.Version
+	initialHeight int64
+	providers     map[lightprovider.Provider]string
+}
+
+// NewLightClientStateProvider creates a new StateProvider using a light client and RPC clients.
+func NewLightClientStateProvider(
+	ctx context.Context,
+	chainID string,
+	version sm.Version,
+	initialHeight int64,
+	servers []string,
+	trustOptions light.TrustOptions,
+	logger log.Logger,
+) (StateProvider, error) {
+	if len(servers) < 2 {
+		return nil, fmt.Errorf("at least 2 RPC servers are required, got %v", len(servers))
+	}
+
+	providers := make([]lightprovider.Provider, 0, len(servers))
+	providerRemotes := make(map[lightprovider.Provider]string)
+	for _, server := range servers {
+		client, err := rpcClient(server)
+		if err != nil {
+			return nil, fmt.Errorf("failed to set up RPC client: %w", err)
+		}
+		provider := lighthttp.NewWithClient(chainID, client)
+		providers = append(providers, provider)
+		// We store the RPC addresses keyed by provider, so we can find the address of the primary
+		// provider used by the light client and use it to fetch consensus parameters.
+		providerRemotes[provider] = server
+	}
+
+	lc, err := light.NewClient(ctx, chainID, trustOptions, providers[0], providers[1:],
+		lightdb.New(dbm.NewMemDB()), light.Logger(logger))
+	if err != nil {
+		return nil, err
+	}
+	return &lightClientStateProvider{
+		lc:            lc,
+		version:       version,
+		initialHeight: initialHeight,
+		providers:     providerRemotes,
+	}, nil
+}
+
+// AppHash implements StateProvider.
+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(ctx, int64(height+1), time.Now())
+	if err != nil {
+		return nil, err
+	}
+	// We also try to fetch the blocks at height H and H+2, since we need these
+	// when building the state while restoring the snapshot. This avoids the race
+	// condition where we try to restore a snapshot before H+2 exists.
+	//
+	// FIXME This is a hack, since we can't add new methods to the interface without
+	// breaking it. We should instead have a Has(ctx, height) method which checks
+	// that the state provider has access to the necessary data for the height.
+	// We piggyback on AppHash() since it's called when adding snapshots to the pool.
+	_, err = s.lc.VerifyLightBlockAtHeight(ctx, int64(height+2), time.Now())
+	if err != nil {
+		return nil, err
+	}
+	_, err = s.lc.VerifyLightBlockAtHeight(ctx, int64(height), time.Now())
+	if err != nil {
+		return nil, err
+	}
+	return header.AppHash, nil
+}
+
+// Commit implements StateProvider.
+func (s *lightClientStateProvider) Commit(ctx context.Context, height uint64) (*types.Commit, error) {
+	s.Lock()
+	defer s.Unlock()
+	header, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height), time.Now())
+	if err != nil {
+		return nil, err
+	}
+	return header.Commit, nil
+}
+
+// State implements StateProvider.
+func (s *lightClientStateProvider) State(ctx context.Context, height uint64) (sm.State, error) {
+	s.Lock()
+	defer s.Unlock()
+
+	state := sm.State{
+		ChainID:       s.lc.ChainID(),
+		Version:       s.version,
+		InitialHeight: s.initialHeight,
+	}
+	if state.InitialHeight == 0 {
+		state.InitialHeight = 1
+	}
+
+	// The snapshot height maps onto the state heights as follows:
+	//
+	// height: last block, i.e. the snapshotted height
+	// height+1: current block, i.e. the first block we'll process after the snapshot
+	// height+2: next block, i.e. the second block after the snapshot
+	//
+	// 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(ctx, int64(height), time.Now())
+	if err != nil {
+		return sm.State{}, err
+	}
+	currentLightBlock, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height+1), time.Now())
+	if err != nil {
+		return sm.State{}, err
+	}
+	nextLightBlock, err := s.lc.VerifyLightBlockAtHeight(ctx, int64(height+2), time.Now())
+	if err != nil {
+		return sm.State{}, err
+	}
+
+	state.LastBlockHeight = lastLightBlock.Height
+	state.LastBlockTime = lastLightBlock.Time
+	state.LastBlockID = lastLightBlock.Commit.BlockID
+	state.AppHash = currentLightBlock.AppHash
+	state.LastResultsHash = currentLightBlock.LastResultsHash
+	state.LastValidators = lastLightBlock.ValidatorSet
+	state.Validators = currentLightBlock.ValidatorSet
+	state.NextValidators = nextLightBlock.ValidatorSet
+	state.LastHeightValidatorsChanged = nextLightBlock.Height
+
+	// We'll also need to fetch consensus params via RPC, using light client verification.
+	primaryURL, ok := s.providers[s.lc.Primary()]
+	if !ok || primaryURL == "" {
+		return sm.State{}, fmt.Errorf("could not find address for primary light client provider")
+	}
+	primaryRPC, err := rpcClient(primaryURL)
+	if err != nil {
+		return sm.State{}, fmt.Errorf("unable to create RPC client: %w", err)
+	}
+	rpcclient := lightrpc.NewClient(primaryRPC, s.lc)
+	result, err := rpcclient.ConsensusParams(ctx, &currentLightBlock.Height)
+	if err != nil {
+		return sm.State{}, fmt.Errorf("unable to fetch consensus parameters for height %v: %w",
+			nextLightBlock.Height, err)
+	}
+	state.ConsensusParams = result.ConsensusParams
+	state.LastHeightConsensusParamsChanged = currentLightBlock.Height
+
+	return state, nil
+}
+
+// rpcClient sets up a new RPC client
+func rpcClient(server string) (*rpchttp.HTTP, error) {
+	if !strings.Contains(server, "://") {
+		server = "http://" + server
+	}
+	c, err := rpchttp.New(server)
+	if err != nil {
+		return nil, err
+	}
+	return c, nil
+}

internal/statesync/syncer.go

@@ -0,0 +1,480 @@
+package statesync
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	abci "github.com/tendermint/tendermint/abci/types"
+	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
+	"github.com/tendermint/tendermint/internal/p2p"
+	"github.com/tendermint/tendermint/libs/log"
+	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 (
+	// chunkFetchers is the number of concurrent chunk fetchers to run.
+	chunkFetchers = 4
+	// chunkTimeout is the timeout while waiting for the next chunk from the chunk queue.
+	chunkTimeout = 2 * time.Minute
+	// requestTimeout is the timeout before rerequesting a chunk, possibly from a different peer.
+	chunkRequestTimeout = 10 * time.Second
+	// minimumDiscoveryTime is the lowest allowable time for a
+	// SyncAny discovery time.
+	minimumDiscoveryTime = 5 * time.Second
+)
+
+var (
+	// errAbort is returned by Sync() when snapshot restoration is aborted.
+	errAbort = errors.New("state sync aborted")
+	// errRetrySnapshot is returned by Sync() when the snapshot should be retried.
+	errRetrySnapshot = errors.New("retry snapshot")
+	// errRejectSnapshot is returned by Sync() when the snapshot is rejected.
+	errRejectSnapshot = errors.New("snapshot was rejected")
+	// errRejectFormat is returned by Sync() when the snapshot format is rejected.
+	errRejectFormat = errors.New("snapshot format was rejected")
+	// errRejectSender is returned by Sync() when the snapshot sender is rejected.
+	errRejectSender = errors.New("snapshot sender was rejected")
+	// errVerifyFailed is returned by Sync() when app hash or last height verification fails.
+	errVerifyFailed = errors.New("verification failed")
+	// errTimeout is returned by Sync() when we've waited too long to receive a chunk.
+	errTimeout = errors.New("timed out waiting for chunk")
+	// errNoSnapshots is returned by SyncAny() if no snapshots are found and discovery is disabled.
+	errNoSnapshots = errors.New("no suitable snapshots found")
+)
+
+// syncer runs a state sync against an ABCI app. Use either SyncAny() to automatically attempt to
+// sync all snapshots in the pool (pausing to discover new ones), or Sync() to sync a specific
+// snapshot. Snapshots and chunks are fed via AddSnapshot() and AddChunk() as appropriate.
+type syncer struct {
+	logger        log.Logger
+	stateProvider StateProvider
+	conn          proxy.AppConnSnapshot
+	connQuery     proxy.AppConnQuery
+	snapshots     *snapshotPool
+	snapshotCh    chan<- p2p.Envelope
+	chunkCh       chan<- p2p.Envelope
+	tempDir       string
+
+	mtx    tmsync.RWMutex
+	chunks *chunkQueue
+}
+
+// newSyncer creates a new syncer.
+func newSyncer(
+	logger log.Logger,
+	conn proxy.AppConnSnapshot,
+	connQuery proxy.AppConnQuery,
+	stateProvider StateProvider,
+	snapshotCh, chunkCh chan<- p2p.Envelope,
+	tempDir string,
+) *syncer {
+	return &syncer{
+		logger:        logger,
+		stateProvider: stateProvider,
+		conn:          conn,
+		connQuery:     connQuery,
+		snapshots:     newSnapshotPool(stateProvider),
+		snapshotCh:    snapshotCh,
+		chunkCh:       chunkCh,
+		tempDir:       tempDir,
+	}
+}
+
+// AddChunk adds a chunk to the chunk queue, if any. It returns false if the chunk has already
+// been added to the queue, or an error if there's no sync in progress.
+func (s *syncer) AddChunk(chunk *chunk) (bool, error) {
+	s.mtx.RLock()
+	defer s.mtx.RUnlock()
+	if s.chunks == nil {
+		return false, errors.New("no state sync in progress")
+	}
+	added, err := s.chunks.Add(chunk)
+	if err != nil {
+		return false, err
+	}
+	if added {
+		s.logger.Debug("Added chunk to queue", "height", chunk.Height, "format", chunk.Format,
+			"chunk", chunk.Index)
+	} else {
+		s.logger.Debug("Ignoring duplicate chunk in queue", "height", chunk.Height, "format", chunk.Format,
+			"chunk", chunk.Index)
+	}
+	return added, nil
+}
+
+// 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 p2p.NodeID, snapshot *snapshot) (bool, error) {
+	added, err := s.snapshots.Add(peerID, snapshot)
+	if err != nil {
+		return false, err
+	}
+	if added {
+		s.logger.Info("Discovered new snapshot", "height", snapshot.Height, "format", snapshot.Format,
+			"hash", snapshot.Hash)
+	}
+	return added, nil
+}
+
+// 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 p2p.NodeID) {
+	s.logger.Debug("Requesting snapshots from peer", "peer", peerID)
+	s.snapshotCh <- p2p.Envelope{
+		To:      peerID,
+		Message: &ssproto.SnapshotsRequest{},
+	}
+}
+
+// RemovePeer removes a peer from the pool.
+func (s *syncer) RemovePeer(peerID p2p.NodeID) {
+	s.logger.Debug("Removing peer from sync", "peer", peerID)
+	s.snapshots.RemovePeer(peerID)
+}
+
+// SyncAny tries to sync any of the snapshots in the snapshot pool, waiting to discover further
+// snapshots if none were found and discoveryTime > 0. It returns the latest state and block commit
+// which the caller must use to bootstrap the node.
+func (s *syncer) SyncAny(discoveryTime time.Duration, retryHook func()) (sm.State, *types.Commit, error) {
+	if discoveryTime != 0 && discoveryTime < minimumDiscoveryTime {
+		discoveryTime = 5 * minimumDiscoveryTime
+	}
+
+	if discoveryTime > 0 {
+		s.logger.Info(fmt.Sprintf("Discovering snapshots for %v", discoveryTime))
+		time.Sleep(discoveryTime)
+	}
+
+	// The app may ask us to retry a snapshot restoration, in which case we need to reuse
+	// the snapshot and chunk queue from the previous loop iteration.
+	var (
+		snapshot *snapshot
+		chunks   *chunkQueue
+		err      error
+	)
+	for {
+		// If not nil, we're going to retry restoration of the same snapshot.
+		if snapshot == nil {
+			snapshot = s.snapshots.Best()
+			chunks = nil
+		}
+		if snapshot == nil {
+			if discoveryTime == 0 {
+				return sm.State{}, nil, errNoSnapshots
+			}
+			retryHook()
+			s.logger.Info(fmt.Sprintf("Discovering snapshots for %v", discoveryTime))
+			time.Sleep(discoveryTime)
+			continue
+		}
+		if chunks == nil {
+			chunks, err = newChunkQueue(snapshot, s.tempDir)
+			if err != nil {
+				return sm.State{}, nil, fmt.Errorf("failed to create chunk queue: %w", err)
+			}
+			defer chunks.Close() // in case we forget to close it elsewhere
+		}
+
+		newState, commit, err := s.Sync(snapshot, chunks)
+		switch {
+		case err == nil:
+			return newState, commit, nil
+
+		case errors.Is(err, errAbort):
+			return sm.State{}, nil, err
+
+		case errors.Is(err, errRetrySnapshot):
+			chunks.RetryAll()
+			s.logger.Info("Retrying snapshot", "height", snapshot.Height, "format", snapshot.Format,
+				"hash", snapshot.Hash)
+			continue
+
+		case errors.Is(err, errTimeout):
+			s.snapshots.Reject(snapshot)
+			s.logger.Error("Timed out waiting for snapshot chunks, rejected snapshot",
+				"height", snapshot.Height, "format", snapshot.Format, "hash", snapshot.Hash)
+
+		case errors.Is(err, errRejectSnapshot):
+			s.snapshots.Reject(snapshot)
+			s.logger.Info("Snapshot rejected", "height", snapshot.Height, "format", snapshot.Format,
+				"hash", snapshot.Hash)
+
+		case errors.Is(err, errRejectFormat):
+			s.snapshots.RejectFormat(snapshot.Format)
+			s.logger.Info("Snapshot format rejected", "format", snapshot.Format)
+
+		case errors.Is(err, errRejectSender):
+			s.logger.Info("Snapshot senders rejected", "height", snapshot.Height, "format", snapshot.Format,
+				"hash", snapshot.Hash)
+			for _, peer := range s.snapshots.GetPeers(snapshot) {
+				s.snapshots.RejectPeer(peer)
+				s.logger.Info("Snapshot sender rejected", "peer", peer)
+			}
+
+		default:
+			return sm.State{}, nil, fmt.Errorf("snapshot restoration failed: %w", err)
+		}
+
+		// Discard snapshot and chunks for next iteration
+		err = chunks.Close()
+		if err != nil {
+			s.logger.Error("Failed to clean up chunk queue", "err", err)
+		}
+		snapshot = nil
+		chunks = nil
+	}
+}
+
+// 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(snapshot *snapshot, chunks *chunkQueue) (sm.State, *types.Commit, error) {
+	s.mtx.Lock()
+	if s.chunks != nil {
+		s.mtx.Unlock()
+		return sm.State{}, nil, errors.New("a state sync is already in progress")
+	}
+	s.chunks = chunks
+	s.mtx.Unlock()
+	defer func() {
+		s.mtx.Lock()
+		s.chunks = nil
+		s.mtx.Unlock()
+	}()
+
+	// Offer snapshot to ABCI app.
+	err := s.offerSnapshot(snapshot)
+	if err != nil {
+		return sm.State{}, nil, err
+	}
+
+	// Spawn chunk fetchers. They will terminate when the chunk queue is closed or context canceled.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	for i := int32(0); i < chunkFetchers; i++ {
+		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(pctx, snapshot.Height)
+	if err != nil {
+		return sm.State{}, nil, fmt.Errorf("failed to build new state: %w", err)
+	}
+	commit, err := s.stateProvider.Commit(pctx, snapshot.Height)
+	if err != nil {
+		return sm.State{}, nil, fmt.Errorf("failed to fetch commit: %w", err)
+	}
+
+	// Restore snapshot
+	err = s.applyChunks(chunks)
+	if err != nil {
+		return sm.State{}, nil, err
+	}
+
+	// Verify app and update app version
+	appVersion, err := s.verifyApp(snapshot)
+	if err != nil {
+		return sm.State{}, nil, err
+	}
+	state.Version.Consensus.App = appVersion
+
+	// Done! 🎉
+	s.logger.Info("Snapshot restored", "height", snapshot.Height, "format", snapshot.Format,
+		"hash", snapshot.Hash)
+
+	return state, commit, nil
+}
+
+// offerSnapshot offers a snapshot to the app. It returns various errors depending on the app's
+// response, or nil if the snapshot was accepted.
+func (s *syncer) offerSnapshot(snapshot *snapshot) error {
+	s.logger.Info("Offering snapshot to ABCI app", "height", snapshot.Height,
+		"format", snapshot.Format, "hash", snapshot.Hash)
+	resp, err := s.conn.OfferSnapshotSync(context.Background(), abci.RequestOfferSnapshot{
+		Snapshot: &abci.Snapshot{
+			Height:   snapshot.Height,
+			Format:   snapshot.Format,
+			Chunks:   snapshot.Chunks,
+			Hash:     snapshot.Hash,
+			Metadata: snapshot.Metadata,
+		},
+		AppHash: snapshot.trustedAppHash,
+	})
+	if err != nil {
+		return fmt.Errorf("failed to offer snapshot: %w", err)
+	}
+	switch resp.Result {
+	case abci.ResponseOfferSnapshot_ACCEPT:
+		s.logger.Info("Snapshot accepted, restoring", "height", snapshot.Height,
+			"format", snapshot.Format, "hash", snapshot.Hash)
+		return nil
+	case abci.ResponseOfferSnapshot_ABORT:
+		return errAbort
+	case abci.ResponseOfferSnapshot_REJECT:
+		return errRejectSnapshot
+	case abci.ResponseOfferSnapshot_REJECT_FORMAT:
+		return errRejectFormat
+	case abci.ResponseOfferSnapshot_REJECT_SENDER:
+		return errRejectSender
+	default:
+		return fmt.Errorf("unknown ResponseOfferSnapshot result %v", resp.Result)
+	}
+}
+
+// applyChunks applies chunks to the app. It returns various errors depending on the app's
+// response, or nil once the snapshot is fully restored.
+func (s *syncer) applyChunks(chunks *chunkQueue) error {
+	for {
+		chunk, err := chunks.Next()
+		if err == errDone {
+			return nil
+		} else if err != nil {
+			return fmt.Errorf("failed to fetch chunk: %w", err)
+		}
+
+		resp, err := s.conn.ApplySnapshotChunkSync(context.Background(), abci.RequestApplySnapshotChunk{
+			Index:  chunk.Index,
+			Chunk:  chunk.Chunk,
+			Sender: string(chunk.Sender),
+		})
+		if err != nil {
+			return fmt.Errorf("failed to apply chunk %v: %w", chunk.Index, err)
+		}
+		s.logger.Info("Applied snapshot chunk to ABCI app", "height", chunk.Height,
+			"format", chunk.Format, "chunk", chunk.Index, "total", chunks.Size())
+
+		// Discard and refetch any chunks as requested by the app
+		for _, index := range resp.RefetchChunks {
+			err := chunks.Discard(index)
+			if err != nil {
+				return fmt.Errorf("failed to discard chunk %v: %w", index, err)
+			}
+		}
+
+		// Reject any senders as requested by the app
+		for _, sender := range resp.RejectSenders {
+			if sender != "" {
+				peerID := p2p.NodeID(sender)
+				s.snapshots.RejectPeer(peerID)
+
+				if err := chunks.DiscardSender(peerID); err != nil {
+					return fmt.Errorf("failed to reject sender: %w", err)
+				}
+			}
+		}
+
+		switch resp.Result {
+		case abci.ResponseApplySnapshotChunk_ACCEPT:
+		case abci.ResponseApplySnapshotChunk_ABORT:
+			return errAbort
+		case abci.ResponseApplySnapshotChunk_RETRY:
+			chunks.Retry(chunk.Index)
+		case abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT:
+			return errRetrySnapshot
+		case abci.ResponseApplySnapshotChunk_REJECT_SNAPSHOT:
+			return errRejectSnapshot
+		default:
+			return fmt.Errorf("unknown ResponseApplySnapshotChunk result %v", resp.Result)
+		}
+	}
+}
+
+// fetchChunks requests chunks from peers, receiving allocations from the chunk queue. Chunks
+// will be received from the reactor via syncer.AddChunks() to chunkQueue.Add().
+func (s *syncer) fetchChunks(ctx context.Context, snapshot *snapshot, chunks *chunkQueue) {
+	for {
+		index, err := chunks.Allocate()
+		if err == errDone {
+			// Keep checking until the context is canceled (restore is done), in case any
+			// chunks need to be refetched.
+			select {
+			case <-ctx.Done():
+				return
+			default:
+			}
+			time.Sleep(2 * time.Second)
+			continue
+		}
+		if err != nil {
+			s.logger.Error("Failed to allocate chunk from queue", "err", err)
+			return
+		}
+		s.logger.Info("Fetching snapshot chunk", "height", snapshot.Height,
+			"format", snapshot.Format, "chunk", index, "total", chunks.Size())
+
+		ticker := time.NewTicker(chunkRequestTimeout)
+		defer ticker.Stop()
+		s.requestChunk(snapshot, index)
+		select {
+		case <-chunks.WaitFor(index):
+		case <-ticker.C:
+			s.requestChunk(snapshot, index)
+		case <-ctx.Done():
+			return
+		}
+		ticker.Stop()
+	}
+}
+
+// requestChunk requests a chunk from a peer.
+func (s *syncer) requestChunk(snapshot *snapshot, chunk uint32) {
+	peer := s.snapshots.GetPeer(snapshot)
+	if peer == "" {
+		s.logger.Error("No valid peers found for snapshot", "height", snapshot.Height,
+			"format", snapshot.Format, "hash", snapshot.Hash)
+		return
+	}
+
+	s.logger.Debug(
+		"Requesting snapshot chunk",
+		"height", snapshot.Height,
+		"format", snapshot.Format,
+		"chunk", chunk,
+		"peer", peer,
+	)
+
+	s.chunkCh <- p2p.Envelope{
+		To: peer,
+		Message: &ssproto.ChunkRequest{
+			Height: snapshot.Height,
+			Format: snapshot.Format,
+			Index:  chunk,
+		},
+	}
+}
+
+// verifyApp verifies the sync, checking the app hash and last block height. It returns the
+// app version, which should be returned as part of the initial state.
+func (s *syncer) verifyApp(snapshot *snapshot) (uint64, error) {
+	resp, err := s.connQuery.InfoSync(context.Background(), proxy.RequestInfo)
+	if err != nil {
+		return 0, fmt.Errorf("failed to query ABCI app for appHash: %w", err)
+	}
+
+	if !bytes.Equal(snapshot.trustedAppHash, resp.LastBlockAppHash) {
+		s.logger.Error("appHash verification failed",
+			"expected", snapshot.trustedAppHash,
+			"actual", resp.LastBlockAppHash)
+		return 0, errVerifyFailed
+	}
+
+	if uint64(resp.LastBlockHeight) != snapshot.Height {
+		s.logger.Error(
+			"ABCI app reported unexpected last block height",
+			"expected", snapshot.Height,
+			"actual", resp.LastBlockHeight,
+		)
+		return 0, errVerifyFailed
+	}
+
+	s.logger.Info("Verified ABCI app", "height", snapshot.Height, "appHash", snapshot.trustedAppHash)
+	return resp.AppVersion, nil
+}

internal/statesync/syncer_test.go

@@ -0,0 +1,703 @@
+package statesync
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"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/p2p"
+	"github.com/tendermint/tendermint/internal/statesync/mocks"
+	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"
+)
+
+var ctx = context.Background()
+
+func TestSyncer_SyncAny(t *testing.T) {
+	state := sm.State{
+		ChainID: "chain",
+		Version: sm.Version{
+			Consensus: version.Consensus{
+				Block: version.BlockProtocol,
+				App:   0,
+			},
+			Software: version.TMVersion,
+		},
+
+		LastBlockHeight: 1,
+		LastBlockID:     types.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")}},
+
+		ConsensusParams:                  *types.DefaultConsensusParams(),
+		LastHeightConsensusParamsChanged: 1,
+	}
+	commit := &types.Commit{BlockID: types.BlockID{Hash: []byte("blockhash")}}
+
+	chunks := []*chunk{
+		{Height: 1, Format: 1, Index: 0, Chunk: []byte{1, 1, 0}},
+		{Height: 1, Format: 1, Index: 1, Chunk: []byte{1, 1, 1}},
+		{Height: 1, Format: 1, Index: 2, Chunk: []byte{1, 1, 2}},
+	}
+	s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
+
+	stateProvider := &mocks.StateProvider{}
+	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{}
+
+	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
+	_, err := rts.syncer.AddChunk(&chunk{Height: 1, Format: 1, Index: 0, Chunk: []byte{1}})
+	require.Error(t, err)
+
+	// Adding a couple of peers should trigger snapshot discovery messages
+	rts.syncer.AddPeer(peerAID)
+	e := <-rts.snapshotOutCh
+	require.Equal(t, &ssproto.SnapshotsRequest{}, e.Message)
+	require.Equal(t, peerAID, e.To)
+
+	rts.syncer.AddPeer(peerBID)
+	e = <-rts.snapshotOutCh
+	require.Equal(t, &ssproto.SnapshotsRequest{}, e.Message)
+	require.Equal(t, peerBID, e.To)
+
+	// Both peers report back with snapshots. One of them also returns a snapshot we don't want, in
+	// format 2, which will be rejected by the ABCI application.
+	new, err := rts.syncer.AddSnapshot(peerAID, s)
+	require.NoError(t, err)
+	require.True(t, new)
+
+	new, err = rts.syncer.AddSnapshot(peerBID, s)
+	require.NoError(t, err)
+	require.False(t, new)
+
+	s2 := &snapshot{Height: 2, Format: 2, Chunks: 3, Hash: []byte{1}}
+	new, err = rts.syncer.AddSnapshot(peerBID, s2)
+	require.NoError(t, err)
+	require.True(t, new)
+
+	new, err = rts.syncer.AddSnapshot(peerCID, s2)
+	require.NoError(t, err)
+	require.False(t, new)
+
+	// We start a sync, with peers sending back chunks when requested. We first reject the snapshot
+	// with height 2 format 2, and accept the snapshot at height 1.
+	connSnapshot.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: &abci.Snapshot{
+			Height: 2,
+			Format: 2,
+			Chunks: 3,
+			Hash:   []byte{1},
+		},
+		AppHash: []byte("app_hash_2"),
+	}).Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
+	connSnapshot.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: &abci.Snapshot{
+			Height:   s.Height,
+			Format:   s.Format,
+			Chunks:   s.Chunks,
+			Hash:     s.Hash,
+			Metadata: s.Metadata,
+		},
+		AppHash: []byte("app_hash"),
+	}).Times(2).Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ACCEPT}, nil)
+
+	chunkRequests := make(map[uint32]int)
+	chunkRequestsMtx := tmsync.Mutex{}
+
+	var wg sync.WaitGroup
+	wg.Add(4)
+
+	go func() {
+		for e := range rts.chunkOutCh {
+			msg, ok := e.Message.(*ssproto.ChunkRequest)
+			assert.True(t, ok)
+
+			assert.EqualValues(t, 1, msg.Height)
+			assert.EqualValues(t, 1, msg.Format)
+			assert.LessOrEqual(t, msg.Index, uint32(len(chunks)))
+
+			added, err := rts.syncer.AddChunk(chunks[msg.Index])
+			assert.NoError(t, err)
+			assert.True(t, added)
+
+			chunkRequestsMtx.Lock()
+			chunkRequests[msg.Index]++
+			chunkRequestsMtx.Unlock()
+
+			wg.Done()
+		}
+	}()
+
+	// The first time we're applying chunk 2 we tell it to retry the snapshot and discard chunk 1,
+	// which should cause it to keep the existing chunk 0 and 2, and restart restoration from
+	// beginning. We also wait for a little while, to exercise the retry logic in fetchChunks().
+	connSnapshot.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+		Index: 2, Chunk: []byte{1, 1, 2},
+	}).Once().Run(func(args mock.Arguments) { time.Sleep(2 * time.Second) }).Return(
+		&abci.ResponseApplySnapshotChunk{
+			Result:        abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT,
+			RefetchChunks: []uint32{1},
+		}, nil)
+
+	connSnapshot.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+		Index: 0, Chunk: []byte{1, 1, 0},
+	}).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+	connSnapshot.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+		Index: 1, Chunk: []byte{1, 1, 1},
+	}).Times(2).Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+	connSnapshot.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+		Index: 2, Chunk: []byte{1, 1, 2},
+	}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+	connQuery.On("InfoSync", ctx, proxy.RequestInfo).Return(&abci.ResponseInfo{
+		AppVersion:       9,
+		LastBlockHeight:  1,
+		LastBlockAppHash: []byte("app_hash"),
+	}, nil)
+
+	newState, lastCommit, err := rts.syncer.SyncAny(0, func() {})
+	require.NoError(t, err)
+
+	wg.Wait()
+
+	chunkRequestsMtx.Lock()
+	require.Equal(t, map[uint32]int{0: 1, 1: 2, 2: 1}, chunkRequests)
+	chunkRequestsMtx.Unlock()
+
+	// The syncer should have updated the state app version from the ABCI info response.
+	expectState := state
+	expectState.Version.Consensus.App = 9
+
+	require.Equal(t, expectState, newState)
+	require.Equal(t, commit, lastCommit)
+
+	connSnapshot.AssertExpectations(t)
+	connQuery.AssertExpectations(t)
+}
+
+func TestSyncer_SyncAny_noSnapshots(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+	rts := setup(t, nil, nil, stateProvider, 2)
+
+	_, _, err := rts.syncer.SyncAny(0, func() {})
+	require.Equal(t, errNoSnapshots, err)
+}
+
+func TestSyncer_SyncAny_abort(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+	rts := setup(t, nil, nil, stateProvider, 2)
+
+	s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
+	peerID := p2p.NodeID("aa")
+
+	_, err := rts.syncer.AddSnapshot(peerID, s)
+	require.NoError(t, err)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(s), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
+
+	_, _, err = rts.syncer.SyncAny(0, func() {})
+	require.Equal(t, errAbort, err)
+	rts.conn.AssertExpectations(t)
+}
+
+func TestSyncer_SyncAny_reject(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+	rts := setup(t, nil, nil, stateProvider, 2)
+
+	// s22 is tried first, then s12, then s11, then errNoSnapshots
+	s22 := &snapshot{Height: 2, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
+	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 := p2p.NodeID("aa")
+
+	_, err := rts.syncer.AddSnapshot(peerID, s22)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerID, s12)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerID, s11)
+	require.NoError(t, err)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(s12), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
+
+	_, _, err = rts.syncer.SyncAny(0, func() {})
+	require.Equal(t, errNoSnapshots, err)
+	rts.conn.AssertExpectations(t)
+}
+
+func TestSyncer_SyncAny_reject_format(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+	rts := setup(t, nil, nil, stateProvider, 2)
+
+	// s22 is tried first, which reject s22 and s12, then s11 will abort.
+	s22 := &snapshot{Height: 2, Format: 2, Chunks: 3, Hash: []byte{1, 2, 3}}
+	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 := p2p.NodeID("aa")
+
+	_, err := rts.syncer.AddSnapshot(peerID, s22)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerID, s12)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerID, s11)
+	require.NoError(t, err)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(s22), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_FORMAT}, nil)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(s11), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_ABORT}, nil)
+
+	_, _, err = rts.syncer.SyncAny(0, func() {})
+	require.Equal(t, errAbort, err)
+	rts.conn.AssertExpectations(t)
+}
+
+func TestSyncer_SyncAny_reject_sender(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+	rts := setup(t, nil, nil, stateProvider, 2)
+
+	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
+	// errNoSnapshots is returned.
+	sa := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
+	sb := &snapshot{Height: 2, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
+	sc := &snapshot{Height: 3, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
+	sbc := &snapshot{Height: 4, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
+
+	_, err := rts.syncer.AddSnapshot(peerAID, sa)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerBID, sb)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerCID, sc)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerBID, sbc)
+	require.NoError(t, err)
+
+	_, err = rts.syncer.AddSnapshot(peerCID, sbc)
+	require.NoError(t, err)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(sbc), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT_SENDER}, nil)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(sa), AppHash: []byte("app_hash"),
+	}).Once().Return(&abci.ResponseOfferSnapshot{Result: abci.ResponseOfferSnapshot_REJECT}, nil)
+
+	_, _, err = rts.syncer.SyncAny(0, func() {})
+	require.Equal(t, errNoSnapshots, err)
+	rts.conn.AssertExpectations(t)
+}
+
+func TestSyncer_SyncAny_abciError(t *testing.T) {
+	stateProvider := &mocks.StateProvider{}
+	stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+	rts := setup(t, nil, nil, stateProvider, 2)
+
+	errBoom := errors.New("boom")
+	s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}}
+
+	peerID := p2p.NodeID("aa")
+
+	_, err := rts.syncer.AddSnapshot(peerID, s)
+	require.NoError(t, err)
+
+	rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+		Snapshot: toABCI(s), AppHash: []byte("app_hash"),
+	}).Once().Return(nil, errBoom)
+
+	_, _, err = rts.syncer.SyncAny(0, func() {})
+	require.True(t, errors.Is(err, errBoom))
+	rts.conn.AssertExpectations(t)
+}
+
+func TestSyncer_offerSnapshot(t *testing.T) {
+	unknownErr := errors.New("unknown error")
+	boom := errors.New("boom")
+
+	testcases := map[string]struct {
+		result    abci.ResponseOfferSnapshot_Result
+		err       error
+		expectErr error
+	}{
+		"accept":           {abci.ResponseOfferSnapshot_ACCEPT, nil, nil},
+		"abort":            {abci.ResponseOfferSnapshot_ABORT, nil, errAbort},
+		"reject":           {abci.ResponseOfferSnapshot_REJECT, nil, errRejectSnapshot},
+		"reject_format":    {abci.ResponseOfferSnapshot_REJECT_FORMAT, nil, errRejectFormat},
+		"reject_sender":    {abci.ResponseOfferSnapshot_REJECT_SENDER, nil, errRejectSender},
+		"unknown":          {abci.ResponseOfferSnapshot_UNKNOWN, nil, unknownErr},
+		"error":            {0, boom, boom},
+		"unknown non-zero": {9, nil, unknownErr},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			stateProvider := &mocks.StateProvider{}
+			stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+			rts := setup(t, nil, nil, stateProvider, 2)
+
+			s := &snapshot{Height: 1, Format: 1, Chunks: 3, Hash: []byte{1, 2, 3}, trustedAppHash: []byte("app_hash")}
+			rts.conn.On("OfferSnapshotSync", ctx, abci.RequestOfferSnapshot{
+				Snapshot: toABCI(s),
+				AppHash:  []byte("app_hash"),
+			}).Return(&abci.ResponseOfferSnapshot{Result: tc.result}, tc.err)
+
+			err := rts.syncer.offerSnapshot(s)
+			if tc.expectErr == unknownErr {
+				require.Error(t, err)
+			} else {
+				unwrapped := errors.Unwrap(err)
+				if unwrapped != nil {
+					err = unwrapped
+				}
+				require.Equal(t, tc.expectErr, err)
+			}
+		})
+	}
+}
+
+func TestSyncer_applyChunks_Results(t *testing.T) {
+	unknownErr := errors.New("unknown error")
+	boom := errors.New("boom")
+
+	testcases := map[string]struct {
+		result    abci.ResponseApplySnapshotChunk_Result
+		err       error
+		expectErr error
+	}{
+		"accept":           {abci.ResponseApplySnapshotChunk_ACCEPT, nil, nil},
+		"abort":            {abci.ResponseApplySnapshotChunk_ABORT, nil, errAbort},
+		"retry":            {abci.ResponseApplySnapshotChunk_RETRY, nil, nil},
+		"retry_snapshot":   {abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT, nil, errRetrySnapshot},
+		"reject_snapshot":  {abci.ResponseApplySnapshotChunk_REJECT_SNAPSHOT, nil, errRejectSnapshot},
+		"unknown":          {abci.ResponseApplySnapshotChunk_UNKNOWN, nil, unknownErr},
+		"error":            {0, boom, boom},
+		"unknown non-zero": {9, nil, unknownErr},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			stateProvider := &mocks.StateProvider{}
+			stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+			rts := setup(t, nil, nil, stateProvider, 2)
+
+			body := []byte{1, 2, 3}
+			chunks, err := newChunkQueue(&snapshot{Height: 1, Format: 1, Chunks: 1}, "")
+			require.NoError(t, err)
+			_, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: body})
+			require.NoError(t, err)
+
+			rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+				Index: 0, Chunk: body,
+			}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: tc.result}, tc.err)
+			if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
+				rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+					Index: 0, Chunk: body,
+				}).Once().Return(&abci.ResponseApplySnapshotChunk{
+					Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+			}
+
+			err = rts.syncer.applyChunks(chunks)
+			if tc.expectErr == unknownErr {
+				require.Error(t, err)
+			} else {
+				unwrapped := errors.Unwrap(err)
+				if unwrapped != nil {
+					err = unwrapped
+				}
+				require.Equal(t, tc.expectErr, err)
+			}
+
+			rts.conn.AssertExpectations(t)
+		})
+	}
+}
+
+func TestSyncer_applyChunks_RefetchChunks(t *testing.T) {
+	// Discarding chunks via refetch_chunks should work the same for all results
+	testcases := map[string]struct {
+		result abci.ResponseApplySnapshotChunk_Result
+	}{
+		"accept":          {abci.ResponseApplySnapshotChunk_ACCEPT},
+		"abort":           {abci.ResponseApplySnapshotChunk_ABORT},
+		"retry":           {abci.ResponseApplySnapshotChunk_RETRY},
+		"retry_snapshot":  {abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT},
+		"reject_snapshot": {abci.ResponseApplySnapshotChunk_REJECT_SNAPSHOT},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			stateProvider := &mocks.StateProvider{}
+			stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+			rts := setup(t, nil, nil, stateProvider, 2)
+
+			chunks, err := newChunkQueue(&snapshot{Height: 1, Format: 1, Chunks: 3}, "")
+			require.NoError(t, err)
+			added, err := chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: []byte{0}})
+			require.True(t, added)
+			require.NoError(t, err)
+			added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 1, Chunk: []byte{1}})
+			require.True(t, added)
+			require.NoError(t, err)
+			added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 2, Chunk: []byte{2}})
+			require.True(t, added)
+			require.NoError(t, err)
+
+			// The first two chunks are accepted, before the last one asks for 1 to be refetched
+			rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+				Index: 0, Chunk: []byte{0},
+			}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+			rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+				Index: 1, Chunk: []byte{1},
+			}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+			rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+				Index: 2, Chunk: []byte{2},
+			}).Once().Return(&abci.ResponseApplySnapshotChunk{
+				Result:        tc.result,
+				RefetchChunks: []uint32{1},
+			}, nil)
+
+			// Since removing the chunk will cause Next() to block, we spawn a goroutine, then
+			// check the queue contents, and finally close the queue to end the goroutine.
+			// We don't really care about the result of applyChunks, since it has separate test.
+			go func() {
+				rts.syncer.applyChunks(chunks) //nolint:errcheck // purposefully ignore error
+			}()
+
+			time.Sleep(50 * time.Millisecond)
+			require.True(t, chunks.Has(0))
+			require.False(t, chunks.Has(1))
+			require.True(t, chunks.Has(2))
+
+			require.NoError(t, chunks.Close())
+		})
+	}
+}
+
+func TestSyncer_applyChunks_RejectSenders(t *testing.T) {
+	// Banning chunks senders via ban_chunk_senders should work the same for all results
+	testcases := map[string]struct {
+		result abci.ResponseApplySnapshotChunk_Result
+	}{
+		"accept":          {abci.ResponseApplySnapshotChunk_ACCEPT},
+		"abort":           {abci.ResponseApplySnapshotChunk_ABORT},
+		"retry":           {abci.ResponseApplySnapshotChunk_RETRY},
+		"retry_snapshot":  {abci.ResponseApplySnapshotChunk_RETRY_SNAPSHOT},
+		"reject_snapshot": {abci.ResponseApplySnapshotChunk_REJECT_SNAPSHOT},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			stateProvider := &mocks.StateProvider{}
+			stateProvider.On("AppHash", mock.Anything, mock.Anything).Return([]byte("app_hash"), nil)
+
+			rts := setup(t, nil, nil, stateProvider, 2)
+
+			// Set up three peers across two snapshots, and ask for one of them to be banned.
+			// It should be banned from all snapshots.
+			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}
+
+			_, err := rts.syncer.AddSnapshot(peerAID, s1)
+			require.NoError(t, err)
+
+			_, err = rts.syncer.AddSnapshot(peerAID, s2)
+			require.NoError(t, err)
+
+			_, err = rts.syncer.AddSnapshot(peerBID, s1)
+			require.NoError(t, err)
+
+			_, err = rts.syncer.AddSnapshot(peerBID, s2)
+			require.NoError(t, err)
+
+			_, err = rts.syncer.AddSnapshot(peerCID, s1)
+			require.NoError(t, err)
+
+			_, err = rts.syncer.AddSnapshot(peerCID, s2)
+			require.NoError(t, err)
+
+			chunks, err := newChunkQueue(s1, "")
+			require.NoError(t, err)
+
+			added, err := chunks.Add(&chunk{Height: 1, Format: 1, Index: 0, Chunk: []byte{0}, Sender: peerAID})
+			require.True(t, added)
+			require.NoError(t, err)
+
+			added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 1, Chunk: []byte{1}, Sender: peerBID})
+			require.True(t, added)
+			require.NoError(t, err)
+
+			added, err = chunks.Add(&chunk{Height: 1, Format: 1, Index: 2, Chunk: []byte{2}, Sender: peerCID})
+			require.True(t, added)
+			require.NoError(t, err)
+
+			// The first two chunks are accepted, before the last one asks for b sender to be rejected
+			rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+				Index: 0, Chunk: []byte{0}, Sender: "aa",
+			}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+			rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+				Index: 1, Chunk: []byte{1}, Sender: "bb",
+			}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+			rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+				Index: 2, Chunk: []byte{2}, Sender: "cc",
+			}).Once().Return(&abci.ResponseApplySnapshotChunk{
+				Result:        tc.result,
+				RejectSenders: []string{string(peerBID)},
+			}, nil)
+
+			// On retry, the last chunk will be tried again, so we just accept it then.
+			if tc.result == abci.ResponseApplySnapshotChunk_RETRY {
+				rts.conn.On("ApplySnapshotChunkSync", ctx, abci.RequestApplySnapshotChunk{
+					Index: 2, Chunk: []byte{2}, Sender: "cc",
+				}).Once().Return(&abci.ResponseApplySnapshotChunk{Result: abci.ResponseApplySnapshotChunk_ACCEPT}, nil)
+			}
+
+			// We don't really care about the result of applyChunks, since it has separate test.
+			// However, it will block on e.g. retry result, so we spawn a goroutine that will
+			// be shut down when the chunk queue closes.
+			go func() {
+				rts.syncer.applyChunks(chunks) //nolint:errcheck // purposefully ignore error
+			}()
+
+			time.Sleep(50 * time.Millisecond)
+
+			s1peers := rts.syncer.snapshots.GetPeers(s1)
+			require.Len(t, s1peers, 2)
+			require.EqualValues(t, "aa", s1peers[0])
+			require.EqualValues(t, "cc", s1peers[1])
+
+			rts.syncer.snapshots.GetPeers(s1)
+			require.Len(t, s1peers, 2)
+			require.EqualValues(t, "aa", s1peers[0])
+			require.EqualValues(t, "cc", s1peers[1])
+
+			require.NoError(t, chunks.Close())
+		})
+	}
+}
+
+func TestSyncer_verifyApp(t *testing.T) {
+	boom := errors.New("boom")
+	s := &snapshot{Height: 3, Format: 1, Chunks: 5, Hash: []byte{1, 2, 3}, trustedAppHash: []byte("app_hash")}
+
+	testcases := map[string]struct {
+		response  *abci.ResponseInfo
+		err       error
+		expectErr error
+	}{
+		"verified": {&abci.ResponseInfo{
+			LastBlockHeight:  3,
+			LastBlockAppHash: []byte("app_hash"),
+			AppVersion:       9,
+		}, nil, nil},
+		"invalid height": {&abci.ResponseInfo{
+			LastBlockHeight:  5,
+			LastBlockAppHash: []byte("app_hash"),
+			AppVersion:       9,
+		}, nil, errVerifyFailed},
+		"invalid hash": {&abci.ResponseInfo{
+			LastBlockHeight:  3,
+			LastBlockAppHash: []byte("xxx"),
+			AppVersion:       9,
+		}, nil, errVerifyFailed},
+		"error": {nil, boom, boom},
+	}
+	for name, tc := range testcases {
+		tc := tc
+		t.Run(name, func(t *testing.T) {
+			rts := setup(t, nil, nil, nil, 2)
+
+			rts.connQuery.On("InfoSync", ctx, proxy.RequestInfo).Return(tc.response, tc.err)
+			version, err := rts.syncer.verifyApp(s)
+			unwrapped := errors.Unwrap(err)
+			if unwrapped != nil {
+				err = unwrapped
+			}
+
+			require.Equal(t, tc.expectErr, err)
+			if err == nil {
+				require.Equal(t, tc.response.AppVersion, version)
+			}
+		})
+	}
+}
+
+func toABCI(s *snapshot) *abci.Snapshot {
+	return &abci.Snapshot{
+		Height:   s.Height,
+		Format:   s.Format,
+		Chunks:   s.Chunks,
+		Hash:     s.Hash,
+		Metadata: s.Metadata,
+	}
+}