func (s *syncer) Sync(ctx context.Context, 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()
	}()
	hctx, hcancel := context.WithTimeout(ctx, 30*time.Second)
	defer hcancel()
	appHash, err := s.stateProvider.AppHash(hctx, snapshot.Height)
	if err != nil {
		if ctx.Err() != nil {
			return sm.State{}, nil, ctx.Err()
		}
		if err == light.ErrNoWitnesses {
			return sm.State{}, nil,
				fmt.Errorf("failed to get app hash at height %d. No witnesses remaining", snapshot.Height)
		}
		s.logger.Info("failed to get and verify tendermint state. Dropping snapshot and trying again",
			"err", err, "height", snapshot.Height)
		return sm.State{}, nil, errRejectSnapshot
	}
	snapshot.trustedAppHash = appHash
	err = s.offerSnapshot(ctx, snapshot)
	if err != nil {
		return sm.State{}, nil, err
	}
	fetchCtx, cancel := context.WithCancel(ctx)
	defer cancel()
	fetchStartTime := time.Now()
	for i := int32(0); i < s.fetchers; i++ {
		go s.fetchChunks(fetchCtx, snapshot, chunks)
	}
	pctx, pcancel := context.WithTimeout(ctx, 1*time.Minute)
	defer pcancel()
	state, err := s.stateProvider.State(pctx, snapshot.Height)
	if err != nil {
		if ctx.Err() != nil {
			return sm.State{}, nil, ctx.Err()
		}
		if err == light.ErrNoWitnesses {
			return sm.State{}, nil,
				fmt.Errorf("failed to get tendermint state at height %d. No witnesses remaining", snapshot.Height)
		}
		s.logger.Info("failed to get and verify tendermint state. Dropping snapshot and trying again",
			"err", err, "height", snapshot.Height)
		return sm.State{}, nil, errRejectSnapshot
	}
	commit, err := s.stateProvider.Commit(pctx, snapshot.Height)
	if err != nil {
		if ctx.Err() != nil {
			return sm.State{}, nil, ctx.Err()
		}
		if err == light.ErrNoWitnesses {
			return sm.State{}, nil,
				fmt.Errorf("failed to get commit at height %d. No witnesses remaining", snapshot.Height)
		}
		s.logger.Info("failed to get and verify commit. Dropping snapshot and trying again",
			"err", err, "height", snapshot.Height)
		return sm.State{}, nil, errRejectSnapshot
	}
	err = s.applyChunks(ctx, chunks, fetchStartTime)
	if err != nil {
		return sm.State{}, nil, err
	}
	appVersion, err := s.verifyApp(snapshot)
	if err != nil {
		return sm.State{}, nil, err
	}
	state.Version.Consensus.App = appVersion
	s.logger.Info("Snapshot restored", "height", snapshot.Height, "format", snapshot.Format,
		"hash", snapshot.Hash)
	return state, commit, 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.
package statesync
import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"time"
	abci "github.com/tendermint/tendermint/abci/types"
	"github.com/tendermint/tendermint/config"
	tmsync "github.com/tendermint/tendermint/internal/libs/sync"
	"github.com/tendermint/tendermint/internal/p2p"
	"github.com/tendermint/tendermint/internal/proxy"
	sm "github.com/tendermint/tendermint/internal/state"
	"github.com/tendermint/tendermint/libs/log"
	"github.com/tendermint/tendermint/light"
	ssproto "github.com/tendermint/tendermint/proto/tendermint/statesync"
	"github.com/tendermint/tendermint/types"
)
const (
	chunkTimeout = 2 * time.Minute
	minimumDiscoveryTime = 5 * time.Second
)
var (
	errAbort = errors.New("state sync aborted")
	errRetrySnapshot = errors.New("retry snapshot")
	errRejectSnapshot = errors.New("snapshot was rejected")
	errRejectFormat = errors.New("snapshot format was rejected")
	errRejectSender = errors.New("snapshot sender was rejected")
	errVerifyFailed = errors.New("verification with app failed")
	errTimeout = errors.New("timed out waiting for chunk")
	errNoSnapshots = errors.New("no suitable snapshots found")
)
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
	fetchers      int32
	retryTimeout  time.Duration
	mtx     tmsync.RWMutex
	chunks  *chunkQueue
	metrics *Metrics
	avgChunkTime             int64
	lastSyncedSnapshotHeight int64
	processingSnapshot       *snapshot
	closeCh                  <-chan struct{}
}
func newSyncer(
	cfg config.StateSyncConfig,
	logger log.Logger,
	conn proxy.AppConnSnapshot,
	connQuery proxy.AppConnQuery,
	stateProvider StateProvider,
	snapshotCh chan<- p2p.Envelope,
	chunkCh chan<- p2p.Envelope,
	closeCh <-chan struct{},
	tempDir string,
	metrics *Metrics,
) *syncer {
	return &syncer{
		logger:        logger,
		stateProvider: stateProvider,
		conn:          conn,
		connQuery:     connQuery,
		snapshots:     newSnapshotPool(),
		snapshotCh:    snapshotCh,
		chunkCh:       chunkCh,
		tempDir:       tempDir,
		fetchers:      cfg.Fetchers,
		retryTimeout:  cfg.ChunkRequestTimeout,
		metrics:       metrics,
		closeCh:       closeCh,
	}
}
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
}
func (s *syncer) AddSnapshot(peerID types.NodeID, snapshot *snapshot) (bool, error) {
	added, err := s.snapshots.Add(peerID, snapshot)
	if err != nil {
		return false, err
	}
	if added {
		s.metrics.TotalSnapshots.Add(1)
		s.logger.Info("Discovered new snapshot", "height", snapshot.Height, "format", snapshot.Format,
			"hash", snapshot.Hash)
	}
	return added, nil
}
func (s *syncer) AddPeer(peerID types.NodeID) (err error) {
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("panic sending peer snapshot request: %v", r)
		}
	}()
	s.logger.Debug("Requesting snapshots from peer", "peer", peerID)
	msg := p2p.Envelope{
		To:      peerID,
		Message: &ssproto.SnapshotsRequest{},
	}
	select {
	case <-s.closeCh:
	case s.snapshotCh <- msg:
	}
	return err
}
func (s *syncer) RemovePeer(peerID types.NodeID) {
	s.logger.Debug("Removing peer from sync", "peer", peerID)
	s.snapshots.RemovePeer(peerID)
}
func (s *syncer) SyncAny(
	ctx context.Context,
	discoveryTime time.Duration,
	requestSnapshots func(),
) (sm.State, *types.Commit, error) {
	if discoveryTime != 0 && discoveryTime < minimumDiscoveryTime {
		discoveryTime = minimumDiscoveryTime
	}
	if discoveryTime > 0 {
		requestSnapshots()
		s.logger.Info(fmt.Sprintf("Discovering snapshots for %v", discoveryTime))
		time.Sleep(discoveryTime)
	}
	var (
		snapshot *snapshot
		chunks   *chunkQueue
		err      error
	)
	for {
		if snapshot == nil {
			snapshot = s.snapshots.Best()
			chunks = nil
		}
		if snapshot == nil {
			if discoveryTime == 0 {
				return sm.State{}, nil, errNoSnapshots
			}
			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()  
		}
		s.processingSnapshot = snapshot
		s.metrics.SnapshotChunkTotal.Set(float64(snapshot.Chunks))
		newState, commit, err := s.Sync(ctx, snapshot, chunks)
		switch {
		case err == nil:
			s.metrics.SnapshotHeight.Set(float64(snapshot.Height))
			s.lastSyncedSnapshotHeight = int64(snapshot.Height)
			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)
		}
		err = chunks.Close()
		if err != nil {
			s.logger.Error("Failed to clean up chunk queue", "err", err)
		}
		snapshot = nil
		chunks = nil
		s.processingSnapshot = nil
	}
}