added simple wrapper for grpc errors for getting nodeID and address

Author: asmyasnikov

internal/conn/conn.go

@@ -365,7 +365,7 @@ func invoke(
 		defer onTransportError(ctx, err)
 
 		if !useWrapping {
-			return opID, issues, err
+			return opID, issues, withConnInfo(err, nodeID, address)
 		}
 
 		if sentMark.canRetry() {
@@ -530,7 +530,7 @@ func (c *conn) NewStream(
 		}()
 
 		if !useWrapping {
-			return nil, err
+			return nil, withConnInfo(err, c.NodeID(), c.Address())
 		}
 
 		if sentMark.canRetry() {

internal/conn/errors.go

@@ -38,3 +38,34 @@ func IsBadConn(err error, goodConnCodes ...grpcCodes.Code) bool {
 
 	return true
 }
+
+type grpcError struct {
+	err error
+
+	nodeID  uint32
+	address string
+}
+
+func (e *grpcError) Error() string {
+	return e.err.Error()
+}
+
+func (e *grpcError) As(target any) bool {
+	return xerrors.As(e.err, target)
+}
+
+func (e *grpcError) NodeID() uint32 {
+	return e.nodeID
+}
+
+func (e *grpcError) Address() string {
+	return e.address
+}
+
+func withConnInfo(err error, nodeID uint32, address string) error {
+	return &grpcError{
+		err:     err,
+		nodeID:  nodeID,
+		address: address,
+	}
+}

internal/conn/errors_test.go

@@ -108,3 +108,21 @@ func TestIsBadConn(t *testing.T) {
 		})
 	}
 }
+
+func TestGrpcError(t *testing.T) {
+	err := withConnInfo(grpcStatus.Error(grpcCodes.Unavailable, "test"), 123, "test:123")
+	require.Equal(t, `rpc error: code = Unavailable desc = test`, err.Error())
+	var nodeID interface {
+		NodeID() uint32
+	}
+	require.ErrorAs(t, err, &nodeID)
+	require.Equal(t, uint32(123), nodeID.NodeID())
+	var address interface {
+		Address() string
+	}
+	require.ErrorAs(t, err, &address)
+	require.Equal(t, "test:123", address.Address())
+	s, has := grpcStatus.FromError(err)
+	require.True(t, has)
+	require.Equal(t, grpcCodes.Unavailable, s.Code())
+}

internal/conn/grpc_client_stream.go

@@ -59,7 +59,7 @@ func (s *grpcClientStream) CloseSend() (err error) {
 		}
 
 		if !s.wrapping {
-			return err
+			return withConnInfo(err, s.parentConn.NodeID(), s.parentConn.Address())
 		}
 
 		return xerrors.WithStackTrace(xerrors.Transport(
@@ -99,7 +99,7 @@ func (s *grpcClientStream) SendMsg(m interface{}) (err error) {
 		}()
 
 		if !s.wrapping {
-			return err
+			return withConnInfo(err, s.parentConn.NodeID(), s.parentConn.Address())
 		}
 
 		if s.sentMark.canRetry() {
@@ -159,7 +159,7 @@ func (s *grpcClientStream) RecvMsg(m interface{}) (err error) { //nolint:funlen
 		}()
 
 		if !s.wrapping {
-			return err
+			return withConnInfo(err, s.parentConn.NodeID(), s.parentConn.Address())
 		}
 
 		if s.sentMark.canRetry() {

internal/xerrors/transport.go

@@ -134,8 +134,8 @@ func IsTransportError(err error, codes ...grpcCodes.Code) bool {
 	var status *grpcStatus.Status
 	if t := (*transportError)(nil); errors.As(err, &t) {
 		status = t.status
-	} else if t, has := grpcStatus.FromError(err); has {
-		status = t
+	} else if s, has := grpcStatus.FromError(err); has {
+		status = s
 	}
 	if status != nil {
 		if len(codes) == 0 {