func (r *Reactor) fetchLightBlock(height uint64) (*types.LightBlock, error) {
	h := int64(height)
	blockMeta := r.blockStore.LoadBlockMeta(h)
	if blockMeta == nil {
		return nil, nil
	}
	commit := r.blockStore.LoadBlockCommit(h)
	if commit == nil {
		return nil, nil
	}
	vals, err := r.stateStore.LoadValidators(h)
	if err != nil {
		return nil, err
	}
	if vals == nil {
		return nil, nil
	}
	return &types.LightBlock{
		SignedHeader: &types.SignedHeader{
			Header: &blockMeta.Header,
			Commit: commit,
		},
		ValidatorSet: vals,
	}, nil
}
fetchLightBlock works out whether the node has a light block at a particular
height and if so returns it so it can be gossiped to peers
package statesync
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"reflect"
	"runtime/debug"
	"sort"
	"time"
	abci "github.com/tendermint/tendermint/abci/types"
	"github.com/tendermint/tendermint/config"
	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
	"github.com/tendermint/tendermint/internal/p2p"
	"github.com/tendermint/tendermint/internal/proxy"
	sm "github.com/tendermint/tendermint/internal/state"
	"github.com/tendermint/tendermint/internal/store"
	"github.com/tendermint/tendermint/libs/log"
	"github.com/tendermint/tendermint/libs/service"
	"github.com/tendermint/tendermint/light"
	"github.com/tendermint/tendermint/light/provider"
	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
	"github.com/tendermint/tendermint/types"
)
var (
	_ service.Service = (*Reactor)(nil)
	_ p2p.Wrapper     = (*ssproto.Message)(nil)
)
const (
	SnapshotChannel = p2p.ChannelID(0x60)
	ChunkChannel = p2p.ChannelID(0x61)
	LightBlockChannel = p2p.ChannelID(0x62)
	ParamsChannel = p2p.ChannelID(0x63)
	recentSnapshots = 10
	snapshotMsgSize = int(4e6)  
	chunkMsgSize = int(16e6)  
	lightBlockMsgSize = int(1e7)  
	paramMsgSize = int(1e5)  
	lightBlockResponseTimeout = 10 * time.Second
	consensusParamsResponseTimeout = 5 * time.Second
	maxLightBlockRequestRetries = 20
)
func GetChannelDescriptors() []*p2p.ChannelDescriptor {
	return []*p2p.ChannelDescriptor{
		{
			ID:                  SnapshotChannel,
			MessageType:         new(ssproto.Message),
			Priority:            6,
			SendQueueCapacity:   10,
			RecvMessageCapacity: snapshotMsgSize,
			RecvBufferCapacity:  128,
		},
		{
			ID:                  ChunkChannel,
			Priority:            3,
			MessageType:         new(ssproto.Message),
			SendQueueCapacity:   4,
			RecvMessageCapacity: chunkMsgSize,
			RecvBufferCapacity:  128,
		},
		{
			ID:                  LightBlockChannel,
			MessageType:         new(ssproto.Message),
			Priority:            5,
			SendQueueCapacity:   10,
			RecvMessageCapacity: lightBlockMsgSize,
			RecvBufferCapacity:  128,
		},
		{
			ID:                  ParamsChannel,
			MessageType:         new(ssproto.Message),
			Priority:            2,
			SendQueueCapacity:   10,
			RecvMessageCapacity: paramMsgSize,
			RecvBufferCapacity:  128,
		},
	}
}
type Metricer interface {
	TotalSnapshots() int64
	ChunkProcessAvgTime() time.Duration
	SnapshotHeight() int64
	SnapshotChunksCount() int64
	SnapshotChunksTotal() int64
	BackFilledBlocks() int64
	BackFillBlocksTotal() int64
}
type Reactor struct {
	service.BaseService
	chainID       string
	initialHeight int64
	cfg           config.StateSyncConfig
	stateStore    sm.Store
	blockStore    *store.BlockStore
	conn        proxy.AppConnSnapshot
	connQuery   proxy.AppConnQuery
	tempDir     string
	snapshotCh  *p2p.Channel
	chunkCh     *p2p.Channel
	blockCh     *p2p.Channel
	paramsCh    *p2p.Channel
	peerUpdates *p2p.PeerUpdates
	closeCh     chan struct{}
	dispatcher *Dispatcher
	peers      *peerList
	mtx           tmsync.RWMutex
	syncer        *syncer
	providers     map[types.NodeID]*BlockProvider
	stateProvider StateProvider
	metrics            *Metrics
	backfillBlockTotal int64
	backfilledBlocks   int64
}
func NewReactor(
	chainID string,
	initialHeight int64,
	cfg config.StateSyncConfig,
	logger log.Logger,
	conn proxy.AppConnSnapshot,
	connQuery proxy.AppConnQuery,
	snapshotCh, chunkCh, blockCh, paramsCh *p2p.Channel,
	peerUpdates *p2p.PeerUpdates,
	stateStore sm.Store,
	blockStore *store.BlockStore,
	tempDir string,
	ssMetrics *Metrics,
) *Reactor {
	r := &Reactor{
		chainID:       chainID,
		initialHeight: initialHeight,
		cfg:           cfg,
		conn:          conn,
		connQuery:     connQuery,
		snapshotCh:    snapshotCh,
		chunkCh:       chunkCh,
		blockCh:       blockCh,
		paramsCh:      paramsCh,
		peerUpdates:   peerUpdates,
		closeCh:       make(chan struct{}),
		tempDir:       tempDir,
		stateStore:    stateStore,
		blockStore:    blockStore,
		peers:         newPeerList(),
		dispatcher:    NewDispatcher(blockCh.Out),
		providers:     make(map[types.NodeID]*BlockProvider),
		metrics:       ssMetrics,
	}
	r.BaseService = *service.NewBaseService(logger, "StateSync", r)
	return r
}
func (r *Reactor) OnStart() error {
	go r.processSnapshotCh()
	go r.processChunkCh()
	go r.processBlockCh()
	go r.processParamsCh()
	go r.processPeerUpdates()
	return nil
}
func (r *Reactor) OnStop() {
	r.dispatcher.Close()
	<-r.dispatcher.Done()
	close(r.closeCh)
	<-r.peerUpdates.Done()
	<-r.snapshotCh.Done()
	<-r.chunkCh.Done()
	<-r.blockCh.Done()
	<-r.paramsCh.Done()
}
func (r *Reactor) Sync(ctx context.Context) (sm.State, error) {
	if err := r.waitForEnoughPeers(ctx, 2); err != nil {
		return sm.State{}, err
	}
	r.mtx.Lock()
	if r.syncer != nil {
		r.mtx.Unlock()
		return sm.State{}, errors.New("a state sync is already in progress")
	}
	if err := r.initStateProvider(ctx, r.chainID, r.initialHeight); err != nil {
		r.mtx.Unlock()
		return sm.State{}, err
	}
	r.syncer = newSyncer(
		r.cfg,
		r.Logger,
		r.conn,
		r.connQuery,
		r.stateProvider,
		r.snapshotCh.Out,
		r.chunkCh.Out,
		r.snapshotCh.Done(),
		r.tempDir,
		r.metrics,
	)
	r.mtx.Unlock()
	defer func() {
		r.mtx.Lock()
		r.syncer = nil
		r.stateProvider = nil
		r.mtx.Unlock()
	}()
	requestSnapshotsHook := func() {
		msg := p2p.Envelope{
			Broadcast: true,
			Message:   &ssproto.SnapshotsRequest{},
		}
		select {
		case <-ctx.Done():
		case <-r.closeCh:
		case r.snapshotCh.Out <- msg:
		}
	}
	state, commit, err := r.syncer.SyncAny(ctx, r.cfg.DiscoveryTime, requestSnapshotsHook)
	if err != nil {
		return sm.State{}, err
	}
	err = r.stateStore.Bootstrap(state)
	if err != nil {
		return sm.State{}, fmt.Errorf("failed to bootstrap node with new state: %w", err)
	}
	err = r.blockStore.SaveSeenCommit(state.LastBlockHeight, commit)
	if err != nil {
		return sm.State{}, fmt.Errorf("failed to store last seen commit: %w", err)
	}
	err = r.Backfill(ctx, state)
	if err != nil {
		r.Logger.Error("backfill failed. Proceeding optimistically...", "err", err)
	}
	return state, nil
}
func (r *Reactor) Backfill(ctx context.Context, state sm.State) error {
	params := state.ConsensusParams.Evidence
	stopHeight := state.LastBlockHeight - params.MaxAgeNumBlocks
	stopTime := state.LastBlockTime.Add(-params.MaxAgeDuration)
	if stopHeight < state.InitialHeight {
		stopHeight = state.InitialHeight
		stopTime = state.LastBlockTime
	}
	return r.backfill(
		ctx,
		state.ChainID,
		state.LastBlockHeight,
		stopHeight,
		state.InitialHeight,
		state.LastBlockID,
		stopTime,
	)
}
func (r *Reactor) backfill(
	ctx context.Context,
	chainID string,
	startHeight, stopHeight, initialHeight int64,
	trustedBlockID types.BlockID,
	stopTime time.Time,
) error {
	r.Logger.Info("starting backfill process...", "startHeight", startHeight,
		"stopHeight", stopHeight, "stopTime", stopTime, "trustedBlockID", trustedBlockID)
	r.backfillBlockTotal = startHeight - stopHeight + 1
	r.metrics.BackFillBlocksTotal.Set(float64(r.backfillBlockTotal))
	const sleepTime = 1 * time.Second
	var (
		lastValidatorSet *types.ValidatorSet
		lastChangeHeight = startHeight
	)
	queue := newBlockQueue(startHeight, stopHeight, initialHeight, stopTime, maxLightBlockRequestRetries)
	for i := 0; i < int(r.cfg.Fetchers); i++ {
		ctxWithCancel, cancel := context.WithCancel(ctx)
		defer cancel()
		go func() {
			for {
				select {
				case height := <-queue.nextHeight():
					peer := r.peers.Pop(ctx)
					r.Logger.Debug("fetching next block", "height", height, "peer", peer)
					subCtx, cancel := context.WithTimeout(ctxWithCancel, lightBlockResponseTimeout)
					defer cancel()
					lb, err := func() (*types.LightBlock, error) {
						defer cancel()
						return r.dispatcher.LightBlock(subCtx, height, peer)
					}()
					r.peers.Append(peer)
					if errors.Is(err, context.Canceled) {
						return
					}
					if err != nil {
						queue.retry(height)
						if errors.Is(err, errNoConnectedPeers) {
							r.Logger.Info("backfill: no connected peers to fetch light blocks from; sleeping...",
								"sleepTime", sleepTime)
							time.Sleep(sleepTime)
						} else {
							r.Logger.Info("backfill: error with fetching light block",
								"height", height, "err", err)
						}
						continue
					}
					if lb == nil {
						r.Logger.Info("backfill: peer didn't have block, fetching from another peer", "height", height)
						queue.retry(height)
						r.peers.Remove(peer)
						continue
					}
					err = lb.ValidateBasic(chainID)
					if err != nil || lb.Height != height {
						r.Logger.Info("backfill: fetched light block failed validate basic, removing peer...",
							"err", err, "height", height)
						queue.retry(height)
						r.blockCh.Error <- p2p.PeerError{
							NodeID: peer,
							Err:    fmt.Errorf("received invalid light block: %w", err),
						}
						continue
					}
					queue.add(lightBlockResponse{
						block: lb,
						peer:  peer,
					})
					r.Logger.Debug("backfill: added light block to processing queue", "height", height)
				case <-queue.done():
					return
				}
			}
		}()
	}
	for {
		select {
		case <-r.closeCh:
			queue.close()
			return nil
		case <-ctx.Done():
			queue.close()
			return nil
		case resp := <-queue.verifyNext():
			if w, g := trustedBlockID.Hash, resp.block.Hash(); !bytes.Equal(w, g) {
				r.Logger.Info("received invalid light block. header hash doesn't match trusted LastBlockID",
					"trustedHash", w, "receivedHash", g, "height", resp.block.Height)
				r.blockCh.Error <- p2p.PeerError{
					NodeID: resp.peer,
					Err:    fmt.Errorf("received invalid light block. Expected hash %v, got: %v", w, g),
				}
				queue.retry(resp.block.Height)
				continue
			}
			err := r.blockStore.SaveSignedHeader(resp.block.SignedHeader, trustedBlockID)
			if err != nil {
				return err
			}
			if lastValidatorSet != nil && !bytes.Equal(resp.block.Header.ValidatorsHash, resp.block.Header.NextValidatorsHash) {
				err = r.stateStore.SaveValidatorSets(resp.block.Height+1, lastChangeHeight, lastValidatorSet)
				if err != nil {
					return err
				}
				lastChangeHeight = resp.block.Height
			}
			trustedBlockID = resp.block.LastBlockID
			queue.success()
			r.Logger.Info("backfill: verified and stored light block", "height", resp.block.Height)
			lastValidatorSet = resp.block.ValidatorSet
			r.backfilledBlocks++
			r.metrics.BackFilledBlocks.Add(1)
			if resp.block.Height < stopHeight {
				r.backfillBlockTotal++
				r.metrics.BackFillBlocksTotal.Set(float64(r.backfillBlockTotal))
			}
		case <-queue.done():
			if err := queue.error(); err != nil {
				return err
			}
			if err := r.stateStore.SaveValidatorSets(queue.terminal.Height, lastChangeHeight, lastValidatorSet); err != nil {
				return err
			}
			r.Logger.Info("successfully completed backfill process", "endHeight", queue.terminal.Height)
			return nil
		}
	}
}
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.Info(
				"advertising snapshot",
				"height", snapshot.Height,
				"format", snapshot.Format,
				"peer", envelope.From,
			)
			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.Info("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
		}
		logger.Info("added snapshot", "height", msg.Height, "format", msg.Format)
	default:
		return fmt.Errorf("received unknown message: %T", msg)
	}
	return nil
}
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
}
func (r *Reactor) handleLightBlockMessage(envelope p2p.Envelope) error {
	switch msg := envelope.Message.(type) {
	case *ssproto.LightBlockRequest:
		r.Logger.Info("received light block request", "height", msg.Height)
		lb, err := r.fetchLightBlock(msg.Height)
		if err != nil {
			r.Logger.Error("failed to retrieve light block", "err", err, "height", msg.Height)
			return err
		}
		if lb == nil {
			r.blockCh.Out <- p2p.Envelope{
				To: envelope.From,
				Message: &ssproto.LightBlockResponse{
					LightBlock: nil,
				},
			}
			return nil
		}
		lbproto, err := lb.ToProto()
		if err != nil {
			r.Logger.Error("marshaling light block to proto", "err", err)
			return nil
		}
		r.blockCh.Out <- p2p.Envelope{
			To: envelope.From,
			Message: &ssproto.LightBlockResponse{
				LightBlock: lbproto,
			},
		}
	case *ssproto.LightBlockResponse:
		var height int64
		if msg.LightBlock != nil {
			height = msg.LightBlock.SignedHeader.Header.Height
		}
		r.Logger.Info("received light block response", "peer", envelope.From, "height", height)
		if err := r.dispatcher.Respond(msg.LightBlock, envelope.From); err != nil {
			r.Logger.Error("error processing light block response", "err", err, "height", height)
		}
	default:
		return fmt.Errorf("received unknown message: %T", msg)
	}
	return nil
}
func (r *Reactor) handleParamsMessage(envelope p2p.Envelope) error {
	switch msg := envelope.Message.(type) {
	case *ssproto.ParamsRequest:
		r.Logger.Debug("received consensus params request", "height", msg.Height)
		cp, err := r.stateStore.LoadConsensusParams(int64(msg.Height))
		if err != nil {
			r.Logger.Error("failed to fetch requested consensus params", "err", err, "height", msg.Height)
			return nil
		}
		cpproto := cp.ToProto()
		r.paramsCh.Out <- p2p.Envelope{
			To: envelope.From,
			Message: &ssproto.ParamsResponse{
				Height:          msg.Height,
				ConsensusParams: cpproto,
			},
		}
	case *ssproto.ParamsResponse:
		r.mtx.RLock()
		defer r.mtx.RUnlock()
		r.Logger.Debug("received consensus params response", "height", msg.Height)
		cp := types.ConsensusParamsFromProto(msg.ConsensusParams)
		if sp, ok := r.stateProvider.(*stateProviderP2P); ok {
			select {
			case sp.paramsRecvCh <- cp:
			case <-time.After(time.Second):
				return errors.New("failed to send consensus params, stateprovider not ready for response")
			}
		} else {
			r.Logger.Debug("received unexpected params response; using RPC state provider", "peer", envelope.From)
		}
	default:
		return fmt.Errorf("received unknown message: %T", msg)
	}
	return nil
}
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.Error(
				"recovering from processing message panic",
				"err", err,
				"stack", string(debug.Stack()),
			)
		}
	}()
	r.Logger.Debug("received message", "message", reflect.TypeOf(envelope.Message), "peer", envelope.From)
	switch chID {
	case SnapshotChannel:
		err = r.handleSnapshotMessage(envelope)
	case ChunkChannel:
		err = r.handleChunkMessage(envelope)
	case LightBlockChannel:
		err = r.handleLightBlockMessage(envelope)
	case ParamsChannel:
		err = r.handleParamsMessage(envelope)
	default:
		err = fmt.Errorf("unknown channel ID (%d) for envelope (%v)", chID, envelope)
	}
	return err
}
func (r *Reactor) processSnapshotCh() {
	r.processCh(r.snapshotCh, "snapshot")
}
func (r *Reactor) processChunkCh() {
	r.processCh(r.chunkCh, "chunk")
}
func (r *Reactor) processBlockCh() {
	r.processCh(r.blockCh, "light block")
}
func (r *Reactor) processParamsCh() {
	r.processCh(r.paramsCh, "consensus params")
}
func (r *Reactor) processCh(ch *p2p.Channel, chName string) {
	defer ch.Close()
	for {
		select {
		case envelope := <-ch.In:
			if err := r.handleMessage(ch.ID, envelope); err != nil {
				r.Logger.Error(fmt.Sprintf("failed to process %s message", chName),
					"ch_id", ch.ID, "envelope", envelope, "err", err)
				ch.Error <- p2p.PeerError{
					NodeID: envelope.From,
					Err:    err,
				}
			}
		case <-r.closeCh:
			r.Logger.Debug(fmt.Sprintf("stopped listening on %s channel; closing...", chName))
			return
		}
	}
}
func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
	r.Logger.Info("received peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
	switch peerUpdate.Status {
	case p2p.PeerStatusUp:
		r.peers.Append(peerUpdate.NodeID)
	case p2p.PeerStatusDown:
		r.peers.Remove(peerUpdate.NodeID)
	}
	r.mtx.Lock()
	defer r.mtx.Unlock()
	if r.syncer == nil {
		return
	}
	switch peerUpdate.Status {
	case p2p.PeerStatusUp:
		newProvider := NewBlockProvider(peerUpdate.NodeID, r.chainID, r.dispatcher)
		r.providers[peerUpdate.NodeID] = newProvider
		err := r.syncer.AddPeer(peerUpdate.NodeID)
		if err != nil {
			r.Logger.Error("error adding peer to syncer", "error", err)
			return
		}
		if sp, ok := r.stateProvider.(*stateProviderP2P); ok {
			go sp.addProvider(newProvider)
		}
	case p2p.PeerStatusDown:
		delete(r.providers, peerUpdate.NodeID)
		r.syncer.RemovePeer(peerUpdate.NodeID)
	}
	r.Logger.Info("processed peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
}
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
		}
	}
}
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
}