refactor(statesync): implement statesync peer manager

abci/types/application.go

@@ -6,11 +6,25 @@ import (
 	"github.com/tendermint/tendermint/crypto"
 )
 
+// StateSyncer is an interface that handles the state sync protocol
+type StateSyncer interface {
+	// ListSnapshots returns list available snapshots
+	ListSnapshots(context.Context, *RequestListSnapshots) (*ResponseListSnapshots, error)
+	// OfferSnapshot accepts or rejects an offered snapshot to the state synchronization
+	OfferSnapshot(context.Context, *RequestOfferSnapshot) (*ResponseOfferSnapshot, error)
+	// LoadSnapshotChunk loads a chunk of snapshot
+	LoadSnapshotChunk(context.Context, *RequestLoadSnapshotChunk) (*ResponseLoadSnapshotChunk, error)
+	// ApplySnapshotChunk applies a chunk of snapshot
+	ApplySnapshotChunk(context.Context, *RequestApplySnapshotChunk) (*ResponseApplySnapshotChunk, error)
+}
+
 // Application is an interface that enables any finite, deterministic state machine
 // to be driven by a blockchain-based replication engine via the ABCI.
 //
 //go:generate ../../scripts/mockery_generate.sh Application
 type Application interface {
+	StateSyncer
+
 	// Info/Query Connection
 	Info(context.Context, *RequestInfo) (*ResponseInfo, error)    // Return application info
 	Query(context.Context, *RequestQuery) (*ResponseQuery, error) // Query for state
@@ -28,12 +42,6 @@ type Application interface {
 	VerifyVoteExtension(context.Context, *RequestVerifyVoteExtension) (*ResponseVerifyVoteExtension, error)
 	// Deliver the decided block with its txs to the Application
 	FinalizeBlock(context.Context, *RequestFinalizeBlock) (*ResponseFinalizeBlock, error)
-
-	// State Sync Connection
-	ListSnapshots(context.Context, *RequestListSnapshots) (*ResponseListSnapshots, error)                // List available snapshots
-	OfferSnapshot(context.Context, *RequestOfferSnapshot) (*ResponseOfferSnapshot, error)                // Offer a snapshot to the application
-	LoadSnapshotChunk(context.Context, *RequestLoadSnapshotChunk) (*ResponseLoadSnapshotChunk, error)    // Load a snapshot chunk
-	ApplySnapshotChunk(context.Context, *RequestApplySnapshotChunk) (*ResponseApplySnapshotChunk, error) // Apply a shapshot chunk
 }
 
 //-------------------------------------------------------

internal/p2p/channel_params.go

@@ -19,10 +19,27 @@ const (
 	ErrorChannel = ChannelID(0x10)
 	// BlockSyncChannel is a channelStore for blocks and status updates
 	BlockSyncChannel = ChannelID(0x40)
+	// SnapshotChannel exchanges snapshot metadata
+	SnapshotChannel = ChannelID(0x60)
+	// ChunkChannel exchanges chunk contents
+	ChunkChannel = ChannelID(0x61)
+	// LightBlockChannel exchanges light blocks
+	LightBlockChannel = ChannelID(0x62)
+	// ParamsChannel exchanges consensus params
+	ParamsChannel = ChannelID(0x63)
 
 	MempoolChannel = ChannelID(0x30)
 
 	blockMaxMsgSize = 1048576 // 1MB TODO make it configurable
+
+	// snapshotMsgSize is the maximum size of a snapshotResponseMessage
+	snapshotMsgSize = int(4e6) // ~4MB
+	// chunkMsgSize is the maximum size of a chunkResponseMessage
+	chunkMsgSize = int(16e6) // ~16MB
+	// lightBlockMsgSize is the maximum size of a lightBlockResponseMessage
+	lightBlockMsgSize = int(1e7) // ~1MB
+	// paramMsgSize is the maximum size of a paramsResponseMessage
+	paramMsgSize = int(1e5) // ~100kb
 )
 
 // ChannelDescriptors returns a map of all supported descriptors
@@ -52,6 +69,38 @@ func ChannelDescriptors(cfg *config.Config) map[ChannelID]*ChannelDescriptor {
 			RecvBufferCapacity:  128,
 			Name:                "mempool",
 		},
+		SnapshotChannel: {
+			ID:                  SnapshotChannel,
+			Priority:            6,
+			SendQueueCapacity:   10,
+			RecvMessageCapacity: snapshotMsgSize,
+			RecvBufferCapacity:  128,
+			Name:                "snapshot",
+		},
+		ChunkChannel: {
+			ID:                  ChunkChannel,
+			Priority:            3,
+			SendQueueCapacity:   4,
+			RecvMessageCapacity: chunkMsgSize,
+			RecvBufferCapacity:  128,
+			Name:                "chunk",
+		},
+		LightBlockChannel: {
+			ID:                  LightBlockChannel,
+			Priority:            5,
+			SendQueueCapacity:   10,
+			RecvMessageCapacity: lightBlockMsgSize,
+			RecvBufferCapacity:  128,
+			Name:                "light-block",
+		},
+		ParamsChannel: {
+			ID:                  ParamsChannel,
+			Priority:            2,
+			SendQueueCapacity:   10,
+			RecvMessageCapacity: paramMsgSize,
+			RecvBufferCapacity:  128,
+			Name:                "params",
+		},
 	}
 }
 
@@ -65,6 +114,18 @@ func ResolveChannelID(msg proto.Message) ChannelID {
 		*blocksync.StatusRequest,
 		*blocksync.StatusResponse:
 		return BlockSyncChannel
+	case *statesync.ChunkRequest,
+		*statesync.ChunkResponse:
+		return ChunkChannel
+	case *statesync.SnapshotsRequest,
+		*statesync.SnapshotsResponse:
+		return SnapshotChannel
+	case *statesync.ParamsRequest,
+		*statesync.ParamsResponse:
+		return ParamsChannel
+	case *statesync.LightBlockRequest,
+		*statesync.LightBlockResponse:
+		return LightBlockChannel
 	case *consensus.NewRoundStep,
 		*consensus.NewValidBlock,
 		*consensus.Proposal,
@@ -75,15 +136,16 @@ func ResolveChannelID(msg proto.Message) ChannelID {
 		*consensus.VoteSetMaj23,
 		*consensus.VoteSetBits,
 		*consensus.Commit,
-		*consensus.HasCommit,
-		*statesync.SnapshotsRequest,
-		*statesync.SnapshotsResponse,
-		*statesync.ChunkRequest,
-		*statesync.ChunkResponse,
-		*statesync.LightBlockRequest,
-		*statesync.LightBlockResponse,
-		*statesync.ParamsRequest,
-		*statesync.ParamsResponse:
+		*consensus.HasCommit:
+		// TODO: enable these channels when they are implemented
+		//*statesync.SnapshotsRequest,
+		//*statesync.SnapshotsResponse,
+		//*statesync.ChunkRequest,
+		//*statesync.ChunkResponse,
+		//*statesync.LightBlockRequest,
+		//*statesync.LightBlockResponse,
+		//*statesync.ParamsRequest,
+		//*statesync.ParamsResponse:
 	case *p2pproto.PexRequest,
 		*p2pproto.PexResponse,
 		*p2pproto.Echo:

internal/p2p/client/client.go

@@ -1,4 +1,5 @@
 //go:generate ../../../scripts/mockery_generate.sh BlockClient
+//go:generate ../../../scripts/mockery_generate.sh SnapshotClient
 
 package client
 
@@ -19,6 +20,7 @@ import (
 	"github.com/tendermint/tendermint/libs/promise"
 	bcproto "github.com/tendermint/tendermint/proto/tendermint/blocksync"
 	protomem "github.com/tendermint/tendermint/proto/tendermint/mempool"
+	"github.com/tendermint/tendermint/proto/tendermint/statesync"
 	"github.com/tendermint/tendermint/types"
 )
 
@@ -52,6 +54,27 @@ type (
 		// GetSyncStatus requests a block synchronization status from all connected peers
 		GetSyncStatus(ctx context.Context) error
 	}
+	// SnapshotClient defines the methods which must be implemented by snapshot client
+	SnapshotClient interface {
+		GetSnapshots(ctx context.Context, peerID types.NodeID) error
+		GetChunk(
+			ctx context.Context,
+			peerID types.NodeID,
+			height uint64,
+			format uint32,
+			index uint32,
+		) (*promise.Promise[*statesync.ChunkResponse], error)
+		GetParams(
+			ctx context.Context,
+			peerID types.NodeID,
+			height uint64,
+		) (*promise.Promise[*statesync.ParamsResponse], error)
+		GetLightBlock(
+			ctx context.Context,
+			peerID types.NodeID,
+			height uint64,
+		) (*promise.Promise[*statesync.LightBlockResponse], error)
+	}
 	TxSender interface {
 		SendTxs(ctx context.Context, peerID types.NodeID, tx types.Tx) error
 	}
@@ -113,22 +136,68 @@ func New(descriptors map[p2p.ChannelID]*p2p.ChannelDescriptor, creator p2p.Chann
 // if response received in time otherwise reject
 func (c *Client) GetBlock(ctx context.Context, height int64, peerID types.NodeID) (*promise.Promise[*bcproto.BlockResponse], error) {
 	reqID := uuid.NewString()
-	err := c.Send(ctx, p2p.Envelope{
-		Attributes: map[string]string{RequestIDAttribute: reqID},
+	msg := &bcproto.BlockRequest{Height: height}
+	respCh, err := c.sendWithResponse(ctx, reqID, peerID, msg)
+	if err != nil {
+		return nil, err
+	}
+	return newPromise[*bcproto.BlockResponse](ctx, peerID, reqID, respCh, c), nil
+}
+
+// GetChunk requests a chunk from a peer and returns promise.Promise which resolve the result
+func (c *Client) GetChunk(
+	ctx context.Context,
+	peerID types.NodeID,
+	height uint64,
+	version uint32,
+	chunkID []byte,
+) (*promise.Promise[*statesync.ChunkResponse], error) {
+	reqID := uuid.NewString()
+	msg := &statesync.ChunkRequest{Height: height, Version: version, ChunkId: chunkID}
+	respCh, err := c.sendWithResponse(ctx, reqID, peerID, msg)
+	if err != nil {
+		return nil, err
+	}
+	return newPromise[*statesync.ChunkResponse](ctx, peerID, reqID, respCh, c), nil
+}
+
+// GetSnapshots requests snapshots from a peer
+func (c *Client) GetSnapshots(ctx context.Context, peerID types.NodeID) error {
+	return c.Send(ctx, p2p.Envelope{
+		Attributes: map[string]string{RequestIDAttribute: uuid.NewString()},
 		To:         peerID,
-		Message:    &bcproto.BlockRequest{Height: height},
+		Message:    &statesync.SnapshotsRequest{},
 	})
+}
+
+// GetParams returns a promise.Promise which resolve the result if response received in time otherwise reject
+func (c *Client) GetParams(
+	ctx context.Context,
+	peerID types.NodeID,
+	height uint64,
+) (*promise.Promise[*statesync.ParamsResponse], error) {
+	reqID := uuid.NewString()
+	msg := &statesync.ParamsRequest{Height: height}
+	respCh, err := c.sendWithResponse(ctx, reqID, peerID, msg)
 	if err != nil {
-		errSendError := c.Send(ctx, p2p.PeerError{
-			NodeID: peerID,
-			Err:    err,
-		})
-		if errSendError != nil {
-			return nil, multierror.Append(err, errSendError)
-		}
+		return nil, err
 	}
-	respCh := c.addPending(reqID)
-	return newPromise[*bcproto.BlockResponse](ctx, peerID, reqID, respCh, c), nil
+	return newPromise[*statesync.ParamsResponse](ctx, peerID, reqID, respCh, c), nil
+}
+
+// GetLightBlock returns a promise.Promise which resolve the result if response received in time otherwise reject
+func (c *Client) GetLightBlock(
+	ctx context.Context,
+	peerID types.NodeID,
+	height uint64,
+) (*promise.Promise[*statesync.LightBlockResponse], error) {
+	reqID := uuid.NewString()
+	msg := &statesync.LightBlockRequest{Height: height}
+	respCh, err := c.sendWithResponse(ctx, reqID, peerID, msg)
+	if err != nil {
+		return nil, err
+	}
+	return newPromise[*statesync.LightBlockResponse](ctx, peerID, reqID, respCh, c), nil
 }
 
 // GetSyncStatus requests a block synchronization status from all connected peers
@@ -238,6 +307,24 @@ func (c *Client) resolveMessage(ctx context.Context, respID string, res result)
 	return nil
 }
 
+func (c *Client) sendWithResponse(ctx context.Context, reqID string, peerID types.NodeID, msg proto.Message) (chan result, error) {
+	err := c.Send(ctx, p2p.Envelope{
+		Attributes: map[string]string{RequestIDAttribute: reqID},
+		To:         peerID,
+		Message:    msg,
+	})
+	if err != nil {
+		errSendError := c.Send(ctx, p2p.PeerError{
+			NodeID: peerID,
+			Err:    err,
+		})
+		if errSendError != nil {
+			return nil, multierror.Append(err, errSendError)
+		}
+	}
+	return c.addPending(reqID), nil
+}
+
 func (c *Client) addPending(reqID string) chan result {
 	respCh := make(chan result, 1)
 	c.pending.Store(reqID, respCh)