Added retry.Invalid method for mark as invalid concrete object

Author: asmyasnikov

internal/pool/pool_test.go

@@ -849,7 +849,7 @@ func TestPool(t *testing.T) { //nolint:gocyclo
 				var (
 					newItems    atomic.Int64
 					deleteItems atomic.Int64
-					expErr      = xerrors.Retryable(errors.New("expected error"), xerrors.InvalidObject())
+					expErr      = errors.New("expected error")
 				)
 				p := New(rootCtx,
 					WithLimit[*testItem, testItem](1),
@@ -874,7 +874,7 @@ func TestPool(t *testing.T) { //nolint:gocyclo
 				)
 				err := p.With(rootCtx, func(ctx context.Context, testItem *testItem) error {
 					if newItems.Load() < 10 {
-						return expErr
+						return xerrors.Retryable(expErr, xerrors.Invalid(testItem))
 					}
 
 					return nil

internal/xerrors/retryable.go

@@ -14,6 +14,7 @@ type retryableError struct {
 	backoffType backoff.Type
 	code        int32
 	traceID     string
+	invalidObjs []any
 }
 
 func (re *retryableError) Code() int32 {
@@ -81,18 +82,27 @@ func WithName(name string) nameOption {
 	return nameOption(name)
 }
 
-type invalidObjectOption struct{}
+type invalidObjectOption struct {
+	obj any
+}
 
-func (invalidObjectOption) applyToRetryableError(re *retryableError) {
+func (opt invalidObjectOption) applyToRetryableError(re *retryableError) {
+	re.invalidObjs = append(re.invalidObjs, opt.obj)
 }
 
 // InvalidObject deprecated option
 //
-// Deprecated
+// Deprecated: use Invalid instead
 func InvalidObject() invalidObjectOption {
 	return invalidObjectOption{}
 }
 
+func Invalid(obj any) invalidObjectOption {
+	return invalidObjectOption{
+		obj: obj,
+	}
+}
+
 func Retryable(err error, opts ...RetryableErrorOption) error {
 	if err == nil {
 		return nil
@@ -158,3 +168,22 @@ func IsRetryableError(err error) bool {
 
 	return false
 }
+
+func IsValid[T comparable](err error, obj T) bool {
+	if err == nil {
+		return true
+	}
+
+	var e *retryableError
+	if !errors.As(err, &e) {
+		return true
+	}
+
+	for _, o := range e.invalidObjs {
+		if invalidObj, has := o.(T); has && invalidObj == obj {
+			return false
+		}
+	}
+
+	return true
+}

internal/xerrors/retryable_test.go

@@ -71,3 +71,16 @@ func TestUnretryableUnwrap(t *testing.T) {
 	wrapped := Unretryable(test)
 	require.ErrorIs(t, wrapped, test)
 }
+
+func TestIsValid(t *testing.T) {
+	type myType struct{}
+	obj := &myType{}
+	err := Retryable(fmt.Errorf("test"), Invalid(obj))
+	require.False(t, IsValid(err, obj))
+	require.True(t, IsValid(err, &myType{}))
+	var objAsInterface any
+	objAsInterface = obj
+	require.False(t, IsValid(err, objAsInterface))
+	objAsInterface = &myType{}
+	require.True(t, IsValid(err, objAsInterface))
+}

internal/xerrors/session_test.go

@@ -38,12 +38,10 @@ func TestMustDeleteTableOrQuerySession(t *testing.T) {
 			Retryable(
 				Transport(grpcStatus.Error(grpcCodes.Unavailable, "")),
 				WithBackoff(backoff.TypeFast),
-				InvalidObject(),
 			),
 			Retryable(
 				grpcStatus.Error(grpcCodes.Unavailable, ""),
 				WithBackoff(backoff.TypeFast),
-				InvalidObject(),
 			),
 			Transport(grpcStatus.Error(grpcCodes.DataLoss, "")),
 			Transport(grpcStatus.Error(grpcCodes.Unauthenticated, "")),

internal/xsql/conn.go

@@ -118,7 +118,10 @@ func (c *Conn) PrepareContext(ctx context.Context, sql string) (_ driver.Stmt, f
 	}()
 
 	if !c.cc.IsValid() {
-		return nil, xerrors.WithStackTrace(errNotReadyConn)
+		return nil, xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(c),
+			xerrors.Invalid(c.cc),
+		))
 	}
 
 	return &Stmt{

internal/xsql/errors.go

@@ -3,13 +3,11 @@ package xsql
 import (
 	"database/sql/driver"
 	"errors"
-
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 )
 
 var (
 	ErrUnsupported         = driver.ErrSkip
 	errDeprecated          = driver.ErrSkip
 	errWrongQueryProcessor = errors.New("wrong query processor")
-	errNotReadyConn        = xerrors.Retryable(errors.New("conn not ready"), xerrors.InvalidObject())
+	errNotReadyConn        = errors.New("conn not ready")
 )

internal/xsql/stmt.go

@@ -36,7 +36,11 @@ func (stmt *Stmt) QueryContext(ctx context.Context, args []driver.NamedValue) (_
 	}()
 
 	if !stmt.conn.cc.IsValid() {
-		return nil, xerrors.WithStackTrace(errNotReadyConn)
+		return nil, xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(stmt),
+			xerrors.Invalid(stmt.conn),
+			xerrors.Invalid(stmt.conn.cc),
+		))
 	}
 
 	sql, params, err := stmt.conn.toYdb(stmt.sql, args...)
@@ -57,7 +61,11 @@ func (stmt *Stmt) ExecContext(ctx context.Context, args []driver.NamedValue) (_
 	}()
 
 	if !stmt.conn.cc.IsValid() {
-		return nil, xerrors.WithStackTrace(errNotReadyConn)
+		return nil, xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(stmt),
+			xerrors.Invalid(stmt.conn),
+			xerrors.Invalid(stmt.conn.cc),
+		))
 	}
 
 	sql, params, err := stmt.conn.toYdb(stmt.sql, args...)

internal/xsql/tx.go

@@ -144,7 +144,11 @@ func (tx *Tx) PrepareContext(ctx context.Context, sql string) (_ driver.Stmt, fi
 		onDone(finalErr)
 	}()
 	if !tx.conn.cc.IsValid() {
-		return nil, badconn.Map(xerrors.WithStackTrace(errNotReadyConn))
+		return nil, badconn.Map(xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(tx),
+			xerrors.Invalid(tx.conn),
+			xerrors.Invalid(tx.conn.cc),
+		)))
 	}
 
 	return &Stmt{

internal/xsql/xquery/conn.go

@@ -43,11 +43,17 @@ func (c *Conn) Exec(ctx context.Context, sql string, params *params.Params) (
 	result driver.Result, finalErr error,
 ) {
 	if !c.IsValid() {
-		return nil, xerrors.WithStackTrace(errNotReadyConn)
+		return nil, xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(c),
+			xerrors.Invalid(c.session),
+		))
 	}
 
 	if !c.isReady() {
-		return nil, xerrors.WithStackTrace(errNotReadyConn)
+		return nil, xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(c),
+			xerrors.Invalid(c.session),
+		))
 	}
 
 	opts := []options.Execute{
@@ -70,11 +76,10 @@ func (c *Conn) Query(ctx context.Context, sql string, params *params.Params) (
 	result driver.RowsNextResultSet, finalErr error,
 ) {
 	if !c.isReady() {
-		return nil, xerrors.WithStackTrace(errNotReadyConn)
-	}
-
-	if !c.isReady() {
-		return nil, xerrors.WithStackTrace(errNotReadyConn)
+		return nil, xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(c),
+			xerrors.Invalid(c.session),
+		))
 	}
 
 	opts := []options.Execute{
@@ -155,11 +160,17 @@ func (c *Conn) IsValid() bool {
 
 func (c *Conn) Ping(ctx context.Context) (finalErr error) {
 	if !c.isReady() {
-		return xerrors.WithStackTrace(errNotReadyConn)
+		return xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(c),
+			xerrors.Invalid(c.session),
+		))
 	}
 
 	if !c.session.IsAlive() {
-		return xerrors.WithStackTrace(errNotReadyConn)
+		return xerrors.WithStackTrace(xerrors.Retryable(errNotReadyConn,
+			xerrors.Invalid(c),
+			xerrors.Invalid(c.session),
+		))
 	}
 
 	err := c.session.Exec(ctx, "select 1")
@@ -178,7 +189,10 @@ func (c *Conn) BeginTx(ctx context.Context, txOptions driver.TxOptions) (common.
 
 func (c *Conn) Close() (finalErr error) {
 	if !c.closed.CompareAndSwap(false, true) {
-		return xerrors.WithStackTrace(errConnClosedEarly)
+		return xerrors.WithStackTrace(xerrors.Retryable(errConnClosedEarly,
+			xerrors.Invalid(c),
+			xerrors.Invalid(c.session),
+		))
 	}
 
 	defer func() {

internal/xsql/xquery/errors.go

@@ -3,13 +3,11 @@ package xquery
 import (
 	"database/sql/driver"
 	"errors"
-
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 )
 
 var (
 	ErrUnsupported     = driver.ErrSkip
 	errDeprecated      = driver.ErrSkip
-	errConnClosedEarly = xerrors.Retryable(errors.New("conn closed early"), xerrors.InvalidObject())
-	errNotReadyConn    = xerrors.Retryable(errors.New("conn not ready"), xerrors.InvalidObject())
+	errConnClosedEarly = errors.New("conn closed early")
+	errNotReadyConn    = errors.New("conn not ready")
 )