Fixed handle of operational errors in topic streams (backported fix only)

Author: rekby

CHANGELOG.md

@@ -1,3 +1,6 @@
+## v3.55.3
+* Fixed handle of operational errors in topic streams (backported fix only)
+
 ## v3.55.2
 * Fixed init info in topic writer, when autoseq num turned off.
 

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",
 			err,

internal/xerrors/xerrors.go

@@ -76,6 +76,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 {