reproduce error in mock test

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed handle of operational errors in topic streams
 * Fixed topic writer infinite reconnections in some cases
 * Refactored nil on err `internal/grpcwrapper/rawydb/issues.go`, when golangci-lint nilerr enabled
 * Refactored nil on err `internal/grpcwrapper/rawtopic/describe_topic.go`, when golangci-lint nilerr enabled

internal/grpcwrapper/rawtopic/rawtopicreader/rawtopicreader.go

@@ -31,7 +31,9 @@ func (s StreamReader) CloseSend() error {
 func (s StreamReader) Recv() (ServerMessage, error) {
 	grpcMess, err := s.Stream.Recv()
 	if err != nil {
-		err = xerrors.Transport(err)
+		if !xerrors.IsErrorFromServer(err) {
+			err = xerrors.Transport(err)
+		}
 
 		return nil, err
 	}

internal/grpcwrapper/rawtopic/rawtopicwriter/streamwriter.go

@@ -41,7 +41,9 @@ func (w *StreamWriter) Recv() (ServerMessage, error) {
 
 	grpcMsg, err := w.Stream.Recv()
 	if err != nil {
-		err = xerrors.Transport(err)
+		if !xerrors.IsErrorFromServer(err) {
+			err = xerrors.Transport(err)
+		}
 
 		return nil, xerrors.WithStackTrace(xerrors.Wrap(fmt.Errorf(
 			"ydb: failed to read grpc message from writer stream: %w",

internal/topic/topicwriterinternal/writer_grpc_mock_test.go

@@ -63,12 +63,15 @@ func (t *topicWriterOperationUnavailable) StreamWrite(server Ydb_Topic_V1.TopicS
 
 	err = server.Send(&Ydb_Topic.StreamWriteMessage_FromServer{
 		Status: Ydb.StatusIds_SUCCESS,
-		ServerMessage: &Ydb_Topic.StreamWriteMessage_FromServer_InitResponse{InitResponse: &Ydb_Topic.StreamWriteMessage_InitResponse{
-			LastSeqNo:       0,
-			SessionId:       "test",
-			PartitionId:     0,
-			SupportedCodecs: nil,
-		}}})
+		ServerMessage: &Ydb_Topic.StreamWriteMessage_FromServer_InitResponse{
+			InitResponse: &Ydb_Topic.StreamWriteMessage_InitResponse{
+				LastSeqNo:       0,
+				SessionId:       "test",
+				PartitionId:     0,
+				SupportedCodecs: nil,
+			},
+		},
+	})
 	if err != nil {
 		return fmt.Errorf("failed to send init response: %w", err)
 	}
@@ -98,7 +101,8 @@ func (t *topicWriterOperationUnavailable) StreamWrite(server Ydb_Topic_V1.TopicS
 		return errors.New("failed to read messages block")
 	}
 
-	if len(messagesMsg.GetClientMessage().(*Ydb_Topic.StreamWriteMessage_FromClient_WriteRequest).WriteRequest.GetMessages()) == 0 {
+	if len(messagesMsg.GetClientMessage().(*Ydb_Topic.StreamWriteMessage_FromClient_WriteRequest).
+		WriteRequest.GetMessages()) == 0 {
 		return errors.New("received zero messages block")
 	}
 
@@ -119,7 +123,8 @@ func (t *topicWriterOperationUnavailable) StreamWrite(server Ydb_Topic_V1.TopicS
 				PartitionId:     0,
 				WriteStatistics: &Ydb_Topic.StreamWriteMessage_WriteResponse_WriteStatistics{},
 			},
-		}})
+		},
+	})
 
 	if err != nil {
 		return fmt.Errorf("failed to sent write ack: %w", err)

internal/xerrors/xerrors.go

@@ -79,6 +79,12 @@ func As(err error, targets ...interface{}) bool {
 	return false
 }
 
+// IsErrorFromServer return true if err returned from server
+// (opposite to raised internally in sdk)
+func IsErrorFromServer(err error) bool {
+	return IsTransportError(err) || IsOperationError(err)
+}
+
 // Is is a improved proxy to errors.Is
 // This need to single import errors
 func Is(err error, targets ...error) bool {

internal/xtest/ydb_grpc_mocks.go

@@ -50,6 +50,7 @@ type grpcMock struct {
 
 func (m *grpcMock) Close() error {
 	m.grpcServer.Stop()
+
 	return m.listener.Close()
 }
 
@@ -99,7 +100,10 @@ func newMockDiscoveryService(host string, port uint32) *mockDiscoveryService {
 	}
 }
 
-func (m mockDiscoveryService) ListEndpoints(ctx context.Context, request *Ydb_Discovery.ListEndpointsRequest) (*Ydb_Discovery.ListEndpointsResponse, error) {
+func (m mockDiscoveryService) ListEndpoints(
+	ctx context.Context,
+	request *Ydb_Discovery.ListEndpointsRequest,
+) (*Ydb_Discovery.ListEndpointsResponse, error) {
 	res := &Ydb_Discovery.ListEndpointsResult{
 		Endpoints: []*Ydb_Discovery.EndpointInfo{
 			{
@@ -121,12 +125,16 @@ func (m mockDiscoveryService) ListEndpoints(ctx context.Context, request *Ydb_Di
 			Ready:  true,
 			Status: Ydb.StatusIds_SUCCESS,
 			Result: &anypb.Any{},
-		}}
+		},
+	}
 	err := resp.GetOperation().GetResult().MarshalFrom(res)
+
 	return resp, err
 }
 
-func (m mockDiscoveryService) WhoAmI(ctx context.Context, request *Ydb_Discovery.WhoAmIRequest) (*Ydb_Discovery.WhoAmIResponse, error) {
-	//TODO implement me
-	panic("implement me")
+func (m mockDiscoveryService) WhoAmI(
+	ctx context.Context,
+	request *Ydb_Discovery.WhoAmIRequest,
+) (*Ydb_Discovery.WhoAmIResponse, error) {
+	panic("unimplemented")
 }