From 1051fcbd5d368430d9873f7ab09353a85e67a868 Mon Sep 17 00:00:00 2001
From: Yevgeny Pats <16490766+yevgenypats@users.noreply.github.com>
Date: Sat, 17 Dec 2022 20:06:23 +0200
Subject: [PATCH] fix: Use correct error codes

---
 internal/servers/destinations.go | 8 ++++----
 internal/servers/source.go       | 4 ++--
 2 files changed, 6 insertions(+), 6 deletions(-)

diff --git a/internal/servers/destinations.go b/internal/servers/destinations.go
index 515dd9211b..68d525cbe8 100644
--- a/internal/servers/destinations.go
+++ b/internal/servers/destinations.go
@@ -70,11 +70,11 @@ func (s *DestinationServer) Write2(msg pb.Destination_Write2Server) error {
 		if err == io.EOF {
 			return msg.SendAndClose(&pb.Write2_Response{})
 		}
-		return fmt.Errorf("write: failed to receive msg: %w", err)
+		return status.Errorf(codes.Internal, "failed to receive msg: %v", err)
 	}
 	var tables schema.Tables
 	if err := json.Unmarshal(r.Tables, &tables); err != nil {
-		return fmt.Errorf("write: failed to unmarshal tables: %w", err)
+		return status.Errorf(codes.InvalidArgument, "failed to unmarshal tables: %v", err)
 	}
 	sourceName := r.Source
 	syncTime := r.Timestamp.AsTime()
@@ -90,7 +90,7 @@ func (s *DestinationServer) Write2(msg pb.Destination_Write2Server) error {
 			if err == io.EOF {
 				close(resources)
 				if err := eg.Wait(); err != nil {
-					return fmt.Errorf("got EOF. failed to wait for plugin: %w", err)
+					return status.Errorf(codes.Internal, "failed to wait for plugin: %v", err)
 				}
 				return msg.SendAndClose(&pb.Write2_Response{})
 			}
@@ -98,7 +98,7 @@ func (s *DestinationServer) Write2(msg pb.Destination_Write2Server) error {
 			if err := eg.Wait(); err != nil {
 				s.Logger.Error().Err(err).Msg("got error. failed to wait for plugin")
 			}
-			return fmt.Errorf("failed to receive msg: %w", err)
+			return status.Errorf(codes.Internal, "failed to receive msg: %v", err)
 		}
 		var resource schema.DestinationResource
 		if err := json.Unmarshal(r.Resource, &resource); err != nil {
diff --git a/internal/servers/source.go b/internal/servers/source.go
index debf99888e..cee794cc99 100644
--- a/internal/servers/source.go
+++ b/internal/servers/source.go
@@ -114,7 +114,7 @@ func (s *SourceServer) Sync2(req *pb.Sync2_Request, stream pb.Source_Sync2Server
 		destResource := resource.ToDestinationResource()
 		b, err := json.Marshal(destResource)
 		if err != nil {
-			return status.Errorf(codes.InvalidArgument, "failed to marshal resource: %v", err)
+			return status.Errorf(codes.Internal, "failed to marshal resource: %v", err)
 		}
 
 		msg := &pb.Sync2_Response{
@@ -128,7 +128,7 @@ func (s *SourceServer) Sync2(req *pb.Sync2_Request, stream pb.Source_Sync2Server
 			continue
 		}
 		if err := stream.Send(msg); err != nil {
-			return fmt.Errorf("failed to send resource: %v", err)
+			return status.Errorf(codes.Internal, "failed to send resource: %v", err)
 		}
 	}