sync -- handle fatal error

x/sync/client.go

@@ -280,7 +280,7 @@ func (c *client) GetRangeProof(
 // [parseFn] parses the raw response.
 // If the request is unsuccessful or the response can't be parsed,
 // retries the request to a different peer until [ctx] expires.
-// Returns [errAppRequestSendFailed] if we fail to send an AppRequest.
+// Returns [errAppSendFailed] if we fail to send an AppRequest/AppResponse.
 // This should be treated as a fatal error.
 func getAndParse[T any](
 	ctx context.Context,
@@ -301,7 +301,7 @@ func getAndParse[T any](
 			}
 		}
 
-		if errors.Is(err, errAppRequestSendFailed) {
+		if errors.Is(err, errAppSendFailed) {
 			// Failing to send an AppRequest is a fatal error.
 			return nil, err
 		}
@@ -340,7 +340,7 @@ func getAndParse[T any](
 // until the node receives a response, failure notification
 // or [ctx] is canceled.
 // Returns the peer's NodeID and response.
-// Returns [errAppRequestSendFailed] if we failed to send an AppRequest.
+// Returns [errAppSendFailed] if we failed to send an AppRequest/AppResponse.
 // This should be treated as fatal.
 // It's safe to call this method multiple times concurrently.
 func (c *client) get(ctx context.Context, request []byte) (ids.NodeID, []byte, error) {

x/sync/client_test.go

@@ -794,18 +794,18 @@ func TestAppRequestSendFailed(t *testing.T) {
 		gomock.Any(),
 		gomock.Any(),
 		gomock.Any(),
-	).Return(ids.NodeID{}, nil, errAppRequestSendFailed).Times(2)
+	).Return(ids.NodeID{}, nil, errAppSendFailed).Times(2)
 
 	_, err := client.GetChangeProof(
 		context.Background(),
 		&pb.SyncGetChangeProofRequest{},
 		nil, // database is unused
 	)
-	require.ErrorIs(err, errAppRequestSendFailed)
+	require.ErrorIs(err, errAppSendFailed)
 
 	_, err = client.GetRangeProof(
 		context.Background(),
 		&pb.SyncGetRangeProofRequest{},
 	)
-	require.ErrorIs(err, errAppRequestSendFailed)
+	require.ErrorIs(err, errAppSendFailed)
 }

x/sync/network_client.go

@@ -29,9 +29,9 @@ const minRequestHandlingDuration = 100 * time.Millisecond
 var (
 	_ NetworkClient = (*networkClient)(nil)
 
-	errAcquiringSemaphore   = errors.New("error acquiring semaphore")
-	errRequestFailed        = errors.New("request failed")
-	errAppRequestSendFailed = errors.New("failed to send AppRequest")
+	errAcquiringSemaphore = errors.New("error acquiring semaphore")
+	errRequestFailed      = errors.New("request failed")
+	errAppSendFailed      = errors.New("failed to send app message")
 )
 
 // NetworkClient defines ability to send request / response through the Network
@@ -189,6 +189,7 @@ func (c *networkClient) getRequestHandler(requestID uint32) (ResponseHandler, bo
 	return handler, true
 }
 
+// If [errAppSendFailed] is returned this should be considered fatal.
 func (c *networkClient) RequestAny(
 	ctx context.Context,
 	minVersion *version.Application,
@@ -212,6 +213,7 @@ func (c *networkClient) RequestAny(
 	return nodeID, response, err
 }
 
+// If [errAppSendFailed] is returned this should be considered fatal.
 func (c *networkClient) Request(
 	ctx context.Context,
 	nodeID ids.NodeID,
@@ -229,6 +231,7 @@ func (c *networkClient) Request(
 
 // Sends [request] to [nodeID] and returns the response.
 // Returns an error if the request failed or [ctx] is canceled.
+// If [errAppSendFailed] is returned this should be considered fatal.
 // Blocks until a response is received or the [ctx] is canceled fails.
 // Releases active requests semaphore if there was an error in sending the request.
 // Assumes [nodeID] is never [c.myNodeID] since we guarantee
@@ -254,7 +257,13 @@ func (c *networkClient) request(
 	// Send an app request to the peer.
 	if err := c.appSender.SendAppRequest(ctx, nodeIDs, requestID, request); err != nil {
 		c.lock.Unlock()
-		return nil, fmt.Errorf("%w: %s", errAppRequestSendFailed, err)
+		c.log.Fatal(
+			"failed to send app request",
+			zap.Stringer("nodeID", nodeID),
+			zap.Int("requestLen", len(request)),
+			zap.Error(err),
+		)
+		return nil, fmt.Errorf("%w: %s", errAppSendFailed, err)
 	}
 
 	handler := newResponseHandler()

x/sync/network_server.go

@@ -58,7 +58,7 @@ func NewNetworkServer(appSender common.AppSender, db DB, log logging.Logger) *Ne
 }
 
 // AppRequest is called by avalanchego -> VM when there is an incoming AppRequest from a peer.
-// Never returns errors as they are considered fatal.
+// Returns a non-nil error iff we fail to send an app message. This is a fatal error.
 // Sends a response back to the sender if length of response returned by the handler > 0.
 func (s *NetworkServer) AppRequest(
 	ctx context.Context,
@@ -121,14 +121,20 @@ func (s *NetworkServer) AppRequest(
 		return nil
 	}
 
-	if err != nil && !isTimeout(err) {
-		// log unexpected errors instead of returning them, since they are fatal.
-		s.log.Warn(
-			"unexpected error handling AppRequest",
-			zap.Stringer("nodeID", nodeID),
-			zap.Uint32("requestID", requestID),
-			zap.Error(err),
-		)
+	if err != nil {
+		if errors.Is(err, errAppSendFailed) {
+			return err
+		}
+
+		if !isTimeout(err) {
+			// log unexpected errors instead of returning them, since they are fatal.
+			s.log.Warn(
+				"unexpected error handling AppRequest",
+				zap.Stringer("nodeID", nodeID),
+				zap.Uint32("requestID", requestID),
+				zap.Error(err),
+			)
+		}
 	}
 	return nil
 }
@@ -141,6 +147,7 @@ func maybeBytesToMaybe(mb *pb.MaybeBytes) maybe.Maybe[[]byte] {
 }
 
 // Generates a change proof and sends it to [nodeID].
+// If [errAppSendFailed] is returned, this should be considered fatal.
 func (s *NetworkServer) HandleChangeProofRequest(
 	ctx context.Context,
 	nodeID ids.NodeID,
@@ -212,8 +219,17 @@ func (s *NetworkServer) HandleChangeProofRequest(
 				return err
 			}
 
-			// TODO handle this fatal error
-			return s.appSender.SendAppResponse(ctx, nodeID, requestID, proofBytes)
+			if err := s.appSender.SendAppResponse(ctx, nodeID, requestID, proofBytes); err != nil {
+				s.log.Fatal(
+					"failed to send app response",
+					zap.Stringer("nodeID", nodeID),
+					zap.Uint32("requestID", requestID),
+					zap.Int("responseLen", len(proofBytes)),
+					zap.Error(err),
+				)
+				return fmt.Errorf("%w: %s", errAppSendFailed, err)
+			}
+			return nil
 		}
 
 		// We generated a change proof. See if it's small enough.
@@ -227,8 +243,17 @@ func (s *NetworkServer) HandleChangeProofRequest(
 		}
 
 		if len(proofBytes) < bytesLimit {
-			// TODO handle this fatal error
-			return s.appSender.SendAppResponse(ctx, nodeID, requestID, proofBytes)
+			if err := s.appSender.SendAppResponse(ctx, nodeID, requestID, proofBytes); err != nil {
+				s.log.Fatal(
+					"failed to send app response",
+					zap.Stringer("nodeID", nodeID),
+					zap.Uint32("requestID", requestID),
+					zap.Int("responseLen", len(proofBytes)),
+					zap.Error(err),
+				)
+				return fmt.Errorf("%w: %s", errAppSendFailed, err)
+			}
+			return nil
 		}
 
 		// The proof was too large. Try to shrink it.
@@ -238,7 +263,7 @@ func (s *NetworkServer) HandleChangeProofRequest(
 }
 
 // Generates a range proof and sends it to [nodeID].
-// TODO danlaine how should we handle context cancellation?
+// If [errAppSendFailed] is returned, this should be considered fatal.
 func (s *NetworkServer) HandleRangeProofRequest(
 	ctx context.Context,
 	nodeID ids.NodeID,
@@ -276,8 +301,17 @@ func (s *NetworkServer) HandleRangeProofRequest(
 	if err != nil {
 		return err
 	}
-	// TODO handle this fatal error
-	return s.appSender.SendAppResponse(ctx, nodeID, requestID, proofBytes)
+	if err := s.appSender.SendAppResponse(ctx, nodeID, requestID, proofBytes); err != nil {
+		s.log.Fatal(
+			"failed to send app response",
+			zap.Stringer("nodeID", nodeID),
+			zap.Uint32("requestID", requestID),
+			zap.Int("responseLen", len(proofBytes)),
+			zap.Error(err),
+		)
+		return fmt.Errorf("%w: %s", errAppSendFailed, err)
+	}
+	return nil
 }
 
 // Get the range proof specified by [req].

x/sync/network_server_test.go

@@ -337,3 +337,112 @@ func Test_Server_GetChangeProof(t *testing.T) {
 		})
 	}
 }
+
+// Test that AppRequest returns a non-nil error if we fail to send
+// an AppRequest or AppResponse.
+func TestAppRequestErrAppSendFailed(t *testing.T) {
+	startRootID := ids.GenerateTestID()
+	endRootID := ids.GenerateTestID()
+
+	type test struct {
+		name        string
+		request     *pb.Request
+		handlerFunc func(*gomock.Controller) *NetworkServer
+		expectedErr error
+	}
+
+	tests := []test{
+		{
+			name: "GetChangeProof",
+			request: &pb.Request{
+				Message: &pb.Request_ChangeProofRequest{
+					ChangeProofRequest: &pb.SyncGetChangeProofRequest{
+						StartRootHash: startRootID[:],
+						EndRootHash:   endRootID[:],
+						StartKey:      []byte{1},
+						EndKey:        &pb.MaybeBytes{Value: []byte{2}},
+						KeyLimit:      100,
+						BytesLimit:    100,
+					},
+				},
+			},
+			handlerFunc: func(ctrl *gomock.Controller) *NetworkServer {
+				sender := common.NewMockSender(ctrl)
+				sender.EXPECT().SendAppResponse(
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+				).Return(errAppSendFailed).AnyTimes()
+
+				db := merkledb.NewMockMerkleDB(ctrl)
+				db.EXPECT().GetChangeProof(
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+				).Return(&merkledb.ChangeProof{}, nil).Times(1)
+
+				return NewNetworkServer(sender, db, logging.NoLog{})
+			},
+			expectedErr: errAppSendFailed,
+		},
+		{
+			name: "GetRangeProof",
+			request: &pb.Request{
+				Message: &pb.Request_RangeProofRequest{
+					RangeProofRequest: &pb.SyncGetRangeProofRequest{
+						RootHash:   endRootID[:],
+						StartKey:   []byte{1},
+						EndKey:     &pb.MaybeBytes{Value: []byte{2}},
+						KeyLimit:   100,
+						BytesLimit: 100,
+					},
+				},
+			},
+			handlerFunc: func(ctrl *gomock.Controller) *NetworkServer {
+				sender := common.NewMockSender(ctrl)
+				sender.EXPECT().SendAppResponse(
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+				).Return(errAppSendFailed).AnyTimes()
+
+				db := merkledb.NewMockMerkleDB(ctrl)
+				db.EXPECT().GetRangeProofAtRoot(
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+					gomock.Any(),
+				).Return(&merkledb.RangeProof{}, nil).Times(1)
+
+				return NewNetworkServer(sender, db, logging.NoLog{})
+			},
+			expectedErr: errAppSendFailed,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require := require.New(t)
+			ctrl := gomock.NewController(t)
+
+			handler := tt.handlerFunc(ctrl)
+			requestBytes, err := proto.Marshal(tt.request)
+			require.NoError(err)
+
+			err = handler.AppRequest(
+				context.Background(),
+				ids.EmptyNodeID,
+				0,
+				time.Now().Add(10*time.Second),
+				requestBytes,
+			)
+			require.ErrorIs(err, tt.expectedErr)
+		})
+	}
+}