Merge pull request #639 from ydb-platform/fix-retryable

* Fixed default state of `internal/xerrors.retryableError`: it inheri…

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,6 @@
+* Fixed default state of `internal/xerrors.retryableError`: it inherit properties from parent error as possible
+* Marked event `grpc/stats.End` as ignored at observing status of grpc connection
+
 ## v3.42.12
 * Replaced the balancer connection to discovery service from short-lived grpc connection to `internal/conn` lazy connection (revert related changes from `v3.42.6`)
 * Marked as deprecated `trace.Driver.OnBalancerDialEntrypoint` event callback

internal/conn/conn.go

@@ -329,21 +329,20 @@ func (c *conn) Invoke(
 
 	err = cc.Invoke(ctx, method, req, res, append(opts, grpc.Trailer(&md))...)
 	if err != nil {
+		defer func() {
+			c.onTransportError(ctx, err)
+		}()
+
 		if useWrapping {
 			err = xerrors.Transport(err,
 				xerrors.WithAddress(c.Address()),
 			)
 			if sentMark.canRetry() {
-				err = xerrors.Retryable(err,
-					xerrors.WithName("Invoke"),
-					xerrors.WithDeleteSession(),
-				)
+				return xerrors.WithStackTrace(xerrors.Retryable(err))
 			}
-			err = xerrors.WithStackTrace(err)
+			return xerrors.WithStackTrace(err)
 		}
 
-		c.onTransportError(ctx, err)
-
 		return err
 	}
 
@@ -417,18 +416,20 @@ func (c *conn) NewStream(
 
 	s, err = cc.NewStream(ctx, desc, method, opts...)
 	if err != nil {
+		defer func() {
+			c.onTransportError(ctx, err)
+		}()
+
 		if useWrapping {
-			err = xerrors.Retryable(
-				xerrors.Transport(err,
-					xerrors.WithAddress(c.Address()),
-				),
-				xerrors.WithName("NewStream"),
-				xerrors.WithDeleteSession(),
+			err = xerrors.Transport(err,
+				xerrors.WithAddress(c.Address()),
 			)
+			if sentMark.canRetry() {
+				return s, xerrors.WithStackTrace(xerrors.Retryable(err))
+			}
+			return s, xerrors.WithStackTrace(err)
 		}
 
-		c.onTransportError(ctx, err)
-
 		return s, err
 	}
 
@@ -495,6 +496,8 @@ func (statsHandler) HandleRPC(ctx context.Context, rpcStats stats.RPCStats) {
 	switch rpcStats.(type) {
 	case *stats.Begin:
 		getContextMark(ctx).markSafeToRetry()
+	case *stats.End:
+		// if data was sent - markDirty() was called early then stats.End received
 	default:
 		getContextMark(ctx).markDirty()
 	}
@@ -511,7 +514,9 @@ type ctxHandleRPCKey struct{}
 var rpcKey = ctxHandleRPCKey{}
 
 func markContext(ctx context.Context) (context.Context, *modificationMark) {
-	mark := &modificationMark{}
+	mark := &modificationMark{
+		safeToRetry: 1,
+	}
 	return context.WithValue(ctx, rpcKey, mark), mark
 }
 

internal/xerrors/retryable.go

@@ -62,9 +62,16 @@ func WithDeleteSession() RetryableErrorOption {
 }
 
 func Retryable(err error, opts ...RetryableErrorOption) error {
-	re := &retryableError{
-		err:  err,
-		name: "CUSTOM",
+	var (
+		e  Error
+		re = &retryableError{
+			err:  err,
+			name: "CUSTOM",
+		}
+	)
+	if As(err, &e) {
+		re.backoffType = e.BackoffType()
+		re.mustDeleteSession = e.MustDeleteSession()
 	}
 	for _, o := range opts {
 		if o != nil {