revert meta apply in internal/db/database.{Invoke,NewStream}

Author: asmyasnikov

CHANGELOG.md

@@ -3,7 +3,7 @@
 * Refactored traces (start and done struct names have prefix about trace)
 * Replaced `errors.Error`, `errors.Errorf` and `errors.ErrorfSkip` to single `errors.WithStackTrace`
 * Refactored table client options 
-* Added `IsTraceError` function to table retry options struct for check error before submit intermediate/done events of retry operation
+* Declared and implemented interface `errors.isYdbError` for checking ybd/non-ydb errors
 * Fixed double tracing table do events
 * Added `retry.WithFastBackoff` and `retry.WithFastBackoff` options
 * Refactored `table.CreateSession` as retry operation with options
@@ -12,7 +12,6 @@
 * Fixed `recursive` param in `ratelimiter.ListResource`
 * Added counting stream usages for exclude park connection if it in use
 * Added `trace.Driver` events about change stream usage and `conn.Release()` call
-* Fixed bug with non-applying meta headers on direct call `db.Discovery.Discover`
 
 ## 3.14.4
 * Implemented auto-removing `conn.Conn` from `conn.Pool` with counting usages of `conn.Conn`

errors.go

@@ -9,7 +9,7 @@ import (
 )
 
 func IterateByIssues(err error, it func(message string, code uint32, severity uint32)) {
-	var o *errors.OpError
+	var o *errors.OperationError
 	if !errors.As(err, &o) {
 		return
 	}
@@ -80,7 +80,7 @@ func IsOperationError(err error, codes ...int32) bool {
 }
 
 func OperationErrorDescription(err error) Error {
-	var o *errors.OpError
+	var o *errors.OperationError
 	if errors.As(err, &o) {
 		return o
 	}

internal/conn/conn.go

@@ -11,6 +11,7 @@ import (
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/connectivity"
+	grpcStatus "google.golang.org/grpc/status"
 
 	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
 
@@ -186,8 +187,8 @@ func (c *conn) take(ctx context.Context) (cc *grpc.ClientConn, err error) {
 
 	if c.isClosed() {
 		return nil, errors.NewGrpcError(
-			codes.Unavailable,
-			errors.WithMsg("ydb driver conn closed early"),
+			errors.WithStatus(grpcStatus.New(codes.Unavailable, "ydb driver conn closed early")),
+			errors.WithErr(err),
 		)
 	}
 
@@ -312,6 +313,7 @@ func (c *conn) Close(ctx context.Context) (err error) {
 	return err
 }
 
+// invoke have behavior like grpc call
 func (c *conn) invoke(
 	ctx context.Context,
 	method string,
@@ -323,25 +325,34 @@ func (c *conn) invoke(
 	cc, err = c.take(ctx)
 	if err != nil {
 		return errors.NewGrpcError(
-			codes.Unavailable,
-			errors.WithMsg("ydb driver conn take failed"),
+			errors.WithStatus(grpcStatus.New(codes.Unavailable, "ydb driver conn take failed")),
 			errors.WithErr(err),
 		)
 	}
 
 	ctx, err = c.config.Meta().Meta(ctx)
 	if err != nil {
 		return errors.NewGrpcError(
-			codes.Unavailable,
-			errors.WithMsg("ydb driver conn apply meta failed"),
+			errors.WithStatus(grpcStatus.New(codes.Unavailable, "ydb driver conn apply meta failed")),
 			errors.WithErr(err),
 		)
 	}
 
 	c.changeUsages(1)
 	defer c.changeUsages(-1)
 
-	return cc.Invoke(ctx, method, req, res, opts...)
+	err = cc.Invoke(ctx, method, req, res, opts...)
+
+	if err != nil {
+		if s, ok := grpcStatus.FromError(err); ok {
+			return errors.NewGrpcError(
+				errors.WithStatus(s),
+			)
+		}
+		return errors.WithStackTrace(err)
+	}
+
+	return nil
 }
 
 func (c *conn) Invoke(
@@ -400,6 +411,7 @@ func (c *conn) Invoke(
 	return err
 }
 
+// newStream have behavior like grpc call
 func (c *conn) newStream(
 	ctx context.Context,
 	desc *grpc.StreamDesc,
@@ -410,25 +422,27 @@ func (c *conn) newStream(
 	cc, err = c.take(ctx)
 	if err != nil {
 		return nil, errors.NewGrpcError(
-			codes.Unavailable,
-			errors.WithMsg("ydb driver conn take failed"),
-			errors.WithErr(err),
-		)
-	}
-
-	ctx, err = c.config.Meta().Meta(ctx)
-	if err != nil {
-		return nil, errors.NewGrpcError(
-			codes.Unavailable,
-			errors.WithMsg("ydb driver conn apply meta failed"),
+			errors.WithStatus(grpcStatus.New(codes.Unavailable, "ydb driver conn take failed")),
 			errors.WithErr(err),
 		)
 	}
 
 	c.changeStreamUsages(1)
 	defer c.changeStreamUsages(-1)
 
-	return cc.NewStream(ctx, desc, method, opts...)
+	var client grpc.ClientStream
+	client, err = cc.NewStream(ctx, desc, method, opts...)
+
+	if err != nil {
+		if s, ok := grpcStatus.FromError(err); ok {
+			return nil, errors.NewGrpcError(
+				errors.WithStatus(s),
+			)
+		}
+		return nil, errors.WithStackTrace(err)
+	}
+
+	return client, nil
 }
 
 func (c *conn) NewStream(

internal/db/database.go

@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/codes"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/discovery"
@@ -113,18 +112,26 @@ func (db *database) Invoke(
 ) error {
 	cc, err := db.cluster.Get(ctx)
 	if err != nil {
-		return errors.NewGrpcError(
-			codes.Unavailable,
-			errors.WithMsg("ydb driver cluster get failed"),
-			errors.WithErr(err),
-		)
+		return errors.WithStackTrace(err)
 	}
+
+	ctx, err = db.config.Meta().Meta(ctx)
+	if err != nil {
+		return errors.WithStackTrace(err)
+	}
+
 	defer func() {
 		if err != nil && errors.MustPessimizeEndpoint(err) {
 			db.cluster.Pessimize(ctx, cc, err)
 		}
 	}()
-	return cc.Invoke(ctx, method, args, reply, opts...)
+
+	err = cc.Invoke(ctx, method, args, reply, opts...)
+	if err != nil {
+		return errors.WithStackTrace(err)
+	}
+
+	return nil
 }
 
 func (db *database) NewStream(
@@ -135,16 +142,25 @@ func (db *database) NewStream(
 ) (grpc.ClientStream, error) {
 	cc, err := db.cluster.Get(ctx)
 	if err != nil {
-		return nil, errors.NewGrpcError(
-			codes.Unavailable,
-			errors.WithMsg("ydb driver cluster get failed"),
-			errors.WithErr(err),
-		)
+		return nil, errors.WithStackTrace(err)
 	}
+
+	ctx, err = db.config.Meta().Meta(ctx)
+	if err != nil {
+		return nil, errors.WithStackTrace(err)
+	}
+
 	defer func() {
 		if err != nil && errors.MustPessimizeEndpoint(err) {
 			db.cluster.Pessimize(ctx, cc, err)
 		}
 	}()
-	return cc.NewStream(ctx, desc, method, opts...)
+
+	var client grpc.ClientStream
+	client, err = cc.NewStream(ctx, desc, method, opts...)
+	if err != nil {
+		return nil, errors.WithStackTrace(err)
+	}
+
+	return client, nil
 }

internal/errors/errors.go

@@ -1,14 +1,10 @@
 package errors
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"io"
-	"runtime"
-	"strconv"
-	"strings"
 )
 
 func IsTimeoutError(err error) bool {
@@ -78,81 +74,3 @@ func Is(err error, targets ...error) bool {
 func New(text string) error {
 	return WithStackTrace(fmt.Errorf("%w", errors.New(text)), WithSkipDepth(1))
 }
-
-// NewWithIssues returns error which contains child issues
-func NewWithIssues(text string, issues ...error) error {
-	return &errorWithIssues{
-		reason: text,
-		issues: issues,
-	}
-}
-
-type errorWithIssues struct {
-	reason string
-	issues []error
-}
-
-func (e *errorWithIssues) Error() string {
-	var b bytes.Buffer
-	b.WriteString(e.reason)
-	b.WriteString(", issues: [")
-	for i, issue := range e.issues {
-		if i != 0 {
-			b.WriteString(", ")
-		}
-		b.WriteString(issue.Error())
-	}
-	b.WriteString("]")
-	return b.String()
-}
-
-type withStackTraceOptions struct {
-	skipDepth int
-}
-
-type withStackTraceOption func(o *withStackTraceOptions)
-
-func WithSkipDepth(skipDepth int) withStackTraceOption {
-	return func(o *withStackTraceOptions) {
-		o.skipDepth = skipDepth
-	}
-}
-
-// WithStackTrace is a wrapper over original err with file:line identification
-func WithStackTrace(err error, opts ...withStackTraceOption) error {
-	options := withStackTraceOptions{}
-	for _, o := range opts {
-		o(&options)
-	}
-	return &stackError{
-		stackRecord: stackRecord(options.skipDepth + 1),
-		err:         err,
-	}
-}
-
-func stackRecord(depth int) string {
-	function, file, line, _ := runtime.Caller(depth + 1)
-	name := runtime.FuncForPC(function).Name()
-	return name + "(" + fileName(file) + ":" + strconv.Itoa(line) + ")"
-}
-
-func fileName(original string) string {
-	i := strings.LastIndex(original, "/")
-	if i == -1 {
-		return original
-	}
-	return original[i+1:]
-}
-
-type stackError struct {
-	stackRecord string
-	err         error
-}
-
-func (e *stackError) Error() string {
-	return e.err.Error() + " at `" + e.stackRecord + "`"
-}
-
-func (e *stackError) Unwrap() error {
-	return e.err
-}

internal/errors/grpc.go

@@ -3,7 +3,6 @@ package errors
 import (
 	"fmt"
 
-	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
 
@@ -13,9 +12,11 @@ type grpcError struct {
 	err    error
 }
 
+func (e *grpcError) isYdbError() {}
+
 func (e *grpcError) Error() string {
 	if e.err != nil {
-		return fmt.Sprintf("%s: %v", e.status.String(), e.err)
+		return fmt.Sprintf("%s: %v", e.err, e.status.String())
 	}
 	return e.status.String()
 }
@@ -28,32 +29,24 @@ func (e *grpcError) GRPCStatus() *status.Status {
 	return e.status
 }
 
-type grpcErrorOptionsHolder struct {
-	msg string
-	err error
-}
-
-type grpcErrorOption func(h *grpcErrorOptionsHolder)
+type grpcErrorOption func(e *grpcError)
 
-func WithMsg(msg string) grpcErrorOption {
-	return func(h *grpcErrorOptionsHolder) {
-		h.msg = msg
+func WithStatus(s *status.Status) grpcErrorOption {
+	return func(e *grpcError) {
+		e.status = s
 	}
 }
 
 func WithErr(err error) grpcErrorOption {
-	return func(h *grpcErrorOptionsHolder) {
-		h.err = err
+	return func(e *grpcError) {
+		e.err = err
 	}
 }
 
-func NewGrpcError(code codes.Code, opts ...grpcErrorOption) error {
-	h := &grpcErrorOptionsHolder{}
+func NewGrpcError(opts ...grpcErrorOption) error {
+	e := &grpcError{}
 	for _, o := range opts {
-		o(h)
-	}
-	return &grpcError{
-		status: status.New(code, h.msg),
-		err:    h.err,
+		o(e)
 	}
+	return e
 }