migrate x/sync to p2p

Signed-off-by: Joshua Kim <20001595+joshua-kim@users.noreply.github.com>

Author: joshua-kim

network/p2p/p2ptest/client.go

@@ -21,25 +21,19 @@ import (
 
 // NewClient generates a client-server pair and returns the client used to
 // communicate with a server with the specified handler
-func NewClient(
-	t *testing.T,
-	ctx context.Context,
-	handler p2p.Handler,
-	clientNodeID ids.NodeID,
-	serverNodeID ids.NodeID,
-) *p2p.Client {
+func NewClient(t *testing.T, rootCtx context.Context, handler p2p.Handler) *p2p.Client {
 	clientSender := &enginetest.Sender{}
 	serverSender := &enginetest.Sender{}
 
+	clientNodeID := ids.GenerateTestNodeID()
 	clientNetwork, err := p2p.NewNetwork(logging.NoLog{}, clientSender, prometheus.NewRegistry(), "")
 	require.NoError(t, err)
 
+	serverNodeID := ids.GenerateTestNodeID()
 	serverNetwork, err := p2p.NewNetwork(logging.NoLog{}, serverSender, prometheus.NewRegistry(), "")
 	require.NoError(t, err)
 
 	clientSender.SendAppGossipF = func(ctx context.Context, _ common.SendConfig, gossipBytes []byte) error {
-		// Send the request asynchronously to avoid deadlock when the server
-		// sends the response back to the client
 		go func() {
 			require.NoError(t, serverNetwork.AppGossip(ctx, clientNodeID, gossipBytes))
 		}()
@@ -58,8 +52,6 @@ func NewClient(
 	}
 
 	serverSender.SendAppResponseF = func(ctx context.Context, _ ids.NodeID, requestID uint32, responseBytes []byte) error {
-		// Send the request asynchronously to avoid deadlock when the server
-		// sends the response back to the client
 		go func() {
 			require.NoError(t, clientNetwork.AppResponse(ctx, serverNodeID, requestID, responseBytes))
 		}()
@@ -68,8 +60,6 @@ func NewClient(
 	}
 
 	serverSender.SendAppErrorF = func(ctx context.Context, _ ids.NodeID, requestID uint32, errorCode int32, errorMessage string) error {
-		// Send the request asynchronously to avoid deadlock when the server
-		// sends the response back to the client
 		go func() {
 			require.NoError(t, clientNetwork.AppRequestFailed(ctx, serverNodeID, requestID, &common.AppError{
 				Code:    errorCode,
@@ -80,10 +70,10 @@ func NewClient(
 		return nil
 	}
 
-	require.NoError(t, clientNetwork.Connected(ctx, clientNodeID, nil))
-	require.NoError(t, clientNetwork.Connected(ctx, serverNodeID, nil))
-	require.NoError(t, serverNetwork.Connected(ctx, clientNodeID, nil))
-	require.NoError(t, serverNetwork.Connected(ctx, serverNodeID, nil))
+	require.NoError(t, clientNetwork.Connected(rootCtx, clientNodeID, nil))
+	require.NoError(t, clientNetwork.Connected(rootCtx, serverNodeID, nil))
+	require.NoError(t, serverNetwork.Connected(rootCtx, clientNodeID, nil))
+	require.NoError(t, serverNetwork.Connected(rootCtx, serverNodeID, nil))
 
 	require.NoError(t, serverNetwork.AddHandler(0, handler))
 	return clientNetwork.NewClient(0)

network/p2p/p2ptest/client_test.go

@@ -27,7 +27,7 @@ func TestNewClient_AppGossip(t *testing.T) {
 		},
 	}
 
-	client := NewClient(t, ctx, testHandler, ids.GenerateTestNodeID(), ids.GenerateTestNodeID())
+	client := NewClient(t, ctx, testHandler)
 	require.NoError(client.AppGossip(ctx, common.SendConfig{}, []byte("foobar")))
 	<-appGossipChan
 }
@@ -94,7 +94,7 @@ func TestNewClient_AppRequest(t *testing.T) {
 				},
 			}
 
-			client := NewClient(t, ctx, testHandler, ids.GenerateTestNodeID(), ids.GenerateTestNodeID())
+			client := NewClient(t, ctx, testHandler)
 			require.NoError(tt.appRequestF(
 				ctx,
 				client,

x/sync/client_test.go

@@ -38,12 +38,12 @@ func newDefaultDBConfig() merkledb.Config {
 	}
 }
 
-func newFlakyRangeProofHandler(
+func newModifiedRangeProofHandler(
 	t *testing.T,
 	db merkledb.MerkleDB,
 	modifyResponse func(response *merkledb.RangeProof),
 ) p2p.Handler {
-	handler := NewGetRangeProofHandler(logging.NoLog{}, db)
+	handler := NewSyncGetRangeProofHandler(logging.NoLog{}, db)
 
 	c := counter{m: 2}
 	return &p2p.TestHandler{
@@ -74,12 +74,12 @@ func newFlakyRangeProofHandler(
 	}
 }
 
-func newFlakyChangeProofHandler(
+func newModifiedChangeProofHandler(
 	t *testing.T,
 	db merkledb.MerkleDB,
 	modifyResponse func(response *merkledb.ChangeProof),
 ) p2p.Handler {
-	handler := NewGetChangeProofHandler(logging.NoLog{}, db)
+	handler := NewSyncGetChangeProofHandler(logging.NoLog{}, db)
 
 	c := counter{m: 2}
 	return &p2p.TestHandler{
@@ -145,14 +145,3 @@ func (c *counter) Inc() int {
 	c.i++
 	return result
 }
-
-type waitingHandler struct {
-	p2p.NoOpHandler
-	handler         p2p.Handler
-	updatedRootChan chan struct{}
-}
-
-func (w *waitingHandler) AppRequest(ctx context.Context, nodeID ids.NodeID, deadline time.Time, requestBytes []byte) ([]byte, *common.AppError) {
-	<-w.updatedRootChan
-	return w.handler.AppRequest(ctx, nodeID, deadline, requestBytes)
-}

x/sync/manager.go

@@ -41,7 +41,7 @@ var (
 	ErrAlreadyStarted                = errors.New("cannot start a Manager that has already been started")
 	ErrAlreadyClosed                 = errors.New("Manager is closed")
 	ErrNoRangeProofClientProvided    = errors.New("range proof client is a required field of the sync config")
-	ErrNoChangeProofClientProvided   = errors.New("change proof client is a required field of the sync config")
+	ErrNoChangeProofClientProvided   = errors.New("change proofclient is a required field of the sync config")
 	ErrNoDatabaseProvided            = errors.New("sync database is a required field of the sync config")
 	ErrNoLogProvided                 = errors.New("log is a required field of the sync config")
 	ErrZeroWorkLimit                 = errors.New("simultaneous work limit must be greater than 0")
@@ -305,12 +305,7 @@ func (m *Manager) doWork(ctx context.Context, work *workItem) {
 		return
 	}
 
-	select {
-	case <-ctx.Done():
-		m.finishWorkItem()
-		return
-	case <-time.After(waitTime):
-	}
+	<-time.After(waitTime)
 
 	if work.localRootID == ids.Empty {
 		// the keys in this range have not been downloaded, so get all key/values
@@ -373,8 +368,7 @@ func (m *Manager) requestChangeProof(ctx context.Context, work *workItem) {
 		defer m.finishWorkItem()
 
 		if err := m.handleChangeProofResponse(ctx, targetRootID, work, request, responseBytes, err); err != nil {
-			// TODO log responses
-			m.config.Log.Debug("dropping response", zap.Error(err), zap.Stringer("request", request))
+			m.config.Log.Debug("dropping response", zap.Error(err))
 			m.retryWork(work)
 			return
 		}
@@ -431,8 +425,7 @@ func (m *Manager) requestRangeProof(ctx context.Context, work *workItem) {
 		defer m.finishWorkItem()
 
 		if err := m.handleRangeProofResponse(ctx, targetRootID, work, request, responseBytes, appErr); err != nil {
-			// TODO log responses
-			m.config.Log.Debug("dropping response", zap.Error(err), zap.Stringer("request", request))
+			m.config.Log.Debug("dropping response", zap.Error(err))
 			m.retryWork(work)
 			return
 		}
@@ -468,11 +461,10 @@ func (m *Manager) retryWork(work *workItem) {
 	m.workLock.Lock()
 	m.unprocessedWork.Insert(work)
 	m.workLock.Unlock()
-	m.unprocessedWorkCond.Signal()
 }
 
 // Returns an error if we should drop the response
-func (m *Manager) shouldHandleResponse(
+func (m *Manager) handleResponse(
 	bytesLimit uint32,
 	responseBytes []byte,
 	err error,
@@ -507,7 +499,7 @@ func (m *Manager) handleRangeProofResponse(
 	responseBytes []byte,
 	err error,
 ) error {
-	if err := m.shouldHandleResponse(request.BytesLimit, responseBytes, err); err != nil {
+	if err := m.handleResponse(request.BytesLimit, responseBytes, err); err != nil {
 		return err
 	}
 
@@ -558,7 +550,7 @@ func (m *Manager) handleChangeProofResponse(
 	responseBytes []byte,
 	err error,
 ) error {
-	if err := m.shouldHandleResponse(request.BytesLimit, responseBytes, err); err != nil {
+	if err := m.handleResponse(request.BytesLimit, responseBytes, err); err != nil {
 		return err
 	}
 
@@ -614,6 +606,7 @@ func (m *Manager) handleChangeProofResponse(
 
 		m.completeWorkItem(ctx, work, largestHandledKey, targetRootID, changeProof.EndProof)
 	case *pb.SyncGetChangeProofResponse_RangeProof:
+
 		var rangeProof merkledb.RangeProof
 		if err := rangeProof.UnmarshalProto(changeProofResp.RangeProof); err != nil {
 			return err

x/sync/network_server.go

@@ -49,8 +49,8 @@ var (
 	errInvalidBounds        = errors.New("start key is greater than end key")
 	errInvalidRootHash      = fmt.Errorf("root hash must have length %d", hashing.HashLen)
 
-	_ p2p.Handler = (*GetChangeProofHandler)(nil)
-	_ p2p.Handler = (*GetRangeProofHandler)(nil)
+	_ p2p.Handler = (*SyncGetChangeProofHandler)(nil)
+	_ p2p.Handler = (*SyncGetRangeProofHandler)(nil)
 )
 
 func maybeBytesToMaybe(mb *pb.MaybeBytes) maybe.Maybe[[]byte] {
@@ -60,30 +60,30 @@ func maybeBytesToMaybe(mb *pb.MaybeBytes) maybe.Maybe[[]byte] {
 	return maybe.Nothing[[]byte]()
 }
 
-func NewGetChangeProofHandler(log logging.Logger, db DB) *GetChangeProofHandler {
-	return &GetChangeProofHandler{
+func NewSyncGetChangeProofHandler(log logging.Logger, db DB) *SyncGetChangeProofHandler {
+	return &SyncGetChangeProofHandler{
 		log: log,
 		db:  db,
 	}
 }
 
-type GetChangeProofHandler struct {
+type SyncGetChangeProofHandler struct {
 	log logging.Logger
 	db  DB
 }
 
-func (*GetChangeProofHandler) AppGossip(context.Context, ids.NodeID, []byte) {}
+func (*SyncGetChangeProofHandler) AppGossip(context.Context, ids.NodeID, []byte) {}
 
-func (g *GetChangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _ time.Time, requestBytes []byte) ([]byte, *common.AppError) {
-	req := &pb.SyncGetChangeProofRequest{}
-	if err := proto.Unmarshal(requestBytes, req); err != nil {
+func (s *SyncGetChangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _ time.Time, requestBytes []byte) ([]byte, *common.AppError) {
+	request := &pb.SyncGetChangeProofRequest{}
+	if err := proto.Unmarshal(requestBytes, request); err != nil {
 		return nil, &common.AppError{
 			Code:    p2p.ErrUnexpected.Code,
 			Message: fmt.Sprintf("failed to unmarshal request: %s", err),
 		}
 	}
 
-	if err := validateChangeProofRequest(req); err != nil {
+	if err := validateChangeProofRequest(request); err != nil {
 		return nil, &common.AppError{
 			Code:    p2p.ErrUnexpected.Code,
 			Message: fmt.Sprintf("invalid request: %s", err),
@@ -92,21 +92,21 @@ func (g *GetChangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _
 
 	// override limits if they exceed caps
 	var (
-		keyLimit   = min(req.KeyLimit, maxKeyValuesLimit)
-		bytesLimit = min(int(req.BytesLimit), maxByteSizeLimit)
-		start      = maybeBytesToMaybe(req.StartKey)
-		end        = maybeBytesToMaybe(req.EndKey)
+		keyLimit   = min(request.KeyLimit, maxKeyValuesLimit)
+		bytesLimit = min(int(request.BytesLimit), maxByteSizeLimit)
+		start      = maybeBytesToMaybe(request.StartKey)
+		end        = maybeBytesToMaybe(request.EndKey)
 	)
 
-	startRoot, err := ids.ToID(req.StartRootHash)
+	startRoot, err := ids.ToID(request.StartRootHash)
 	if err != nil {
 		return nil, &common.AppError{
 			Code:    p2p.ErrUnexpected.Code,
 			Message: fmt.Sprintf("failed to parse start root hash: %s", err),
 		}
 	}
 
-	endRoot, err := ids.ToID(req.EndRootHash)
+	endRoot, err := ids.ToID(request.EndRootHash)
 	if err != nil {
 		return nil, &common.AppError{
 			Code:    p2p.ErrUnexpected.Code,
@@ -120,7 +120,6 @@ func (g *GetChangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _
 			if !errors.Is(err, merkledb.ErrInsufficientHistory) {
 				// We should only fail to get a change proof if we have insufficient history.
 				// Other errors are unexpected.
-				// TODO define custom errors
 				return nil, &common.AppError{
 					Code:    p2p.ErrUnexpected.Code,
 					Message: fmt.Sprintf("failed to get change proof: %s", err),
@@ -141,11 +140,11 @@ func (g *GetChangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _
 				ctx,
 				g.db,
 				&pb.SyncGetRangeProofRequest{
-					RootHash:   req.EndRootHash,
-					StartKey:   req.StartKey,
-					EndKey:     req.EndKey,
-					KeyLimit:   req.KeyLimit,
-					BytesLimit: req.BytesLimit,
+					RootHash:   request.EndRootHash,
+					StartKey:   request.StartKey,
+					EndKey:     request.EndKey,
+					KeyLimit:   request.KeyLimit,
+					BytesLimit: request.BytesLimit,
 				},
 				func(rangeProof *merkledb.RangeProof) ([]byte, error) {
 					return proto.Marshal(&pb.SyncGetChangeProofResponse{
@@ -192,44 +191,48 @@ func (g *GetChangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _
 	}
 }
 
-func NewGetRangeProofHandler(log logging.Logger, db DB) *GetRangeProofHandler {
-	return &GetRangeProofHandler{
+func (*SyncGetChangeProofHandler) CrossChainAppRequest(context.Context, ids.ID, time.Time, []byte) ([]byte, error) {
+	return nil, nil
+}
+
+func NewSyncGetRangeProofHandler(log logging.Logger, db DB) *SyncGetRangeProofHandler {
+	return &SyncGetRangeProofHandler{
 		log: log,
 		db:  db,
 	}
 }
 
-type GetRangeProofHandler struct {
+type SyncGetRangeProofHandler struct {
 	log logging.Logger
 	db  DB
 }
 
-func (*GetRangeProofHandler) AppGossip(context.Context, ids.NodeID, []byte) {}
+func (*SyncGetRangeProofHandler) AppGossip(context.Context, ids.NodeID, []byte) {}
 
-func (g *GetRangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _ time.Time, requestBytes []byte) ([]byte, *common.AppError) {
-	req := &pb.SyncGetRangeProofRequest{}
-	if err := proto.Unmarshal(requestBytes, req); err != nil {
+func (s *SyncGetRangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _ time.Time, requestBytes []byte) ([]byte, *common.AppError) {
+	request := &pb.SyncGetRangeProofRequest{}
+	if err := proto.Unmarshal(requestBytes, request); err != nil {
 		return nil, &common.AppError{
 			Code:    p2p.ErrUnexpected.Code,
 			Message: fmt.Sprintf("failed to unmarshal request: %s", err),
 		}
 	}
 
-	if err := validateRangeProofRequest(req); err != nil {
+	if err := validateRangeProofRequest(request); err != nil {
 		return nil, &common.AppError{
 			Code:    p2p.ErrUnexpected.Code,
 			Message: fmt.Sprintf("invalid range proof request: %s", err),
 		}
 	}
 
 	// override limits if they exceed caps
-	req.KeyLimit = min(req.KeyLimit, maxKeyValuesLimit)
-	req.BytesLimit = min(req.BytesLimit, maxByteSizeLimit)
+	request.KeyLimit = min(request.KeyLimit, maxKeyValuesLimit)
+	request.BytesLimit = min(request.BytesLimit, maxByteSizeLimit)
 
 	proofBytes, err := getRangeProof(
 		ctx,
-		g.db,
-		req,
+		s.db,
+		request,
 		func(rangeProof *merkledb.RangeProof) ([]byte, error) {
 			return proto.Marshal(rangeProof.ToProto())
 		},
@@ -244,6 +247,10 @@ func (g *GetRangeProofHandler) AppRequest(ctx context.Context, _ ids.NodeID, _ t
 	return proofBytes, nil
 }
 
+func (*SyncGetRangeProofHandler) CrossChainAppRequest(context.Context, ids.ID, time.Time, []byte) ([]byte, error) {
+	return nil, nil
+}
+
 // Get the range proof specified by [req].
 // If the generated proof is too large, the key limit is reduced
 // and the proof is regenerated. This process is repeated until