* Skipped explicit Rollback of transaction on errors (server-side a…

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Skipped explicit `Rollback` of transaction on errors (server-side automatically rolled back transactions on errors)
+
 ## v3.67.1
 * Fixed race of stop internal processes on close topic writer
 * Fixed goroutines leak within topic reader on network problems

internal/table/client.go

@@ -666,9 +666,30 @@ func (c *Client) Do(ctx context.Context, op table.Operation, opts ...table.Optio
 		onDone(attempts, finalErr)
 	}()
 
-	err := do(ctx, c, c.config, op, func(err error) {
-		attempts++
-	}, config.RetryOptions...)
+	err := retryBackoff(ctx, c,
+		func(ctx context.Context, s table.Session) (err error) {
+			attempts++
+
+			err = func() error {
+				if panicCallback := c.config.PanicCallback(); panicCallback != nil {
+					defer func() {
+						if e := recover(); e != nil {
+							panicCallback(e)
+						}
+					}()
+				}
+
+				return op(xcontext.MarkRetryCall(ctx), s)
+			}()
+
+			if err != nil {
+				return xerrors.WithStackTrace(err)
+			}
+
+			return nil
+		},
+		config.RetryOptions...,
+	)
 	if err != nil {
 		return xerrors.WithStackTrace(err)
 	}
@@ -695,26 +716,17 @@ func (c *Client) DoTx(ctx context.Context, op table.TxOperation, opts ...table.O
 		onDone(attempts, finalErr)
 	}()
 
-	return retryBackoff(ctx, c,
-		func(ctx context.Context, s table.Session) (err error) {
+	err := retryBackoff(ctx, c,
+		func(ctx context.Context, s table.Session) (finalErr error) {
 			attempts++
 
 			tx, err := s.BeginTransaction(ctx, config.TxSettings)
 			if err != nil {
 				return xerrors.WithStackTrace(err)
 			}
-
 			defer func() {
-				if err != nil {
-					errRollback := tx.Rollback(ctx)
-					if errRollback != nil {
-						err = xerrors.NewWithIssues("",
-							xerrors.WithStackTrace(err),
-							xerrors.WithStackTrace(errRollback),
-						)
-					} else {
-						err = xerrors.WithStackTrace(err)
-					}
+				if finalErr != nil {
+					_ = tx.Rollback(ctx)
 				}
 			}()
 
@@ -743,6 +755,11 @@ func (c *Client) DoTx(ctx context.Context, op table.TxOperation, opts ...table.O
 		},
 		config.RetryOptions...,
 	)
+	if err != nil {
+		return xerrors.WithStackTrace(err)
+	}
+
+	return nil
 }
 
 func (c *Client) internalPoolGCTick(ctx context.Context, idleThreshold time.Duration) {

internal/table/retry.go

@@ -3,8 +3,6 @@ package table
 import (
 	"context"
 
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/table/config"
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table"
@@ -22,44 +20,6 @@ type SessionProvider interface {
 	Put(ctx context.Context, s *session) (err error)
 }
 
-func do(
-	ctx context.Context,
-	c SessionProvider,
-	config *config.Config,
-	op table.Operation,
-	onAttempt func(err error),
-	opts ...retry.Option,
-) (err error) {
-	return retryBackoff(ctx, c,
-		func(ctx context.Context, s table.Session) (err error) {
-			defer func() {
-				if onAttempt != nil {
-					onAttempt(err)
-				}
-			}()
-
-			err = func() error {
-				if panicCallback := config.PanicCallback(); panicCallback != nil {
-					defer func() {
-						if e := recover(); e != nil {
-							panicCallback(e)
-						}
-					}()
-				}
-
-				return op(xcontext.MarkRetryCall(ctx), s)
-			}()
-
-			if err != nil {
-				return xerrors.WithStackTrace(err)
-			}
-
-			return nil
-		},
-		opts...,
-	)
-}
-
 func retryBackoff(
 	ctx context.Context,
 	p SessionProvider,

internal/table/retry_test.go

@@ -11,7 +11,6 @@ import (
 	grpcCodes "google.golang.org/grpc/codes"
 	grpcStatus "google.golang.org/grpc/status"
 
-	"github.com/ydb-platform/ydb-go-sdk/v3/internal/table/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xrand"
@@ -41,12 +40,10 @@ func TestRetryerBackoffRetryCancelation(t *testing.T) {
 			ctx, cancel := xcontext.WithCancel(context.Background())
 			results := make(chan error)
 			go func() {
-				err := do(ctx, p,
-					config.New(),
+				err := retryBackoff(ctx, p,
 					func(ctx context.Context, _ table.Session) error {
 						return testErr
 					},
-					nil,
 					retry.WithFastBackoff(
 						testutil.BackoffFunc(func(n int) <-chan time.Time {
 							ch := make(chan time.Time)
@@ -103,7 +100,7 @@ func TestRetryerBadSession(t *testing.T) {
 		sessions   []table.Session
 	)
 	ctx, cancel := xcontext.WithCancel(context.Background())
-	err := do(ctx, p, config.New(),
+	err := retryBackoff(ctx, p,
 		func(ctx context.Context, s table.Session) error {
 			sessions = append(sessions, s)
 			i++
@@ -115,7 +112,6 @@ func TestRetryerBadSession(t *testing.T) {
 				xerrors.WithStatusCode(Ydb.StatusIds_BAD_SESSION),
 			)
 		},
-		func(err error) {},
 	)
 	if !xerrors.Is(err, context.Canceled) {
 		t.Errorf("unexpected error: %v", err)
@@ -154,17 +150,13 @@ func TestRetryerSessionClosing(t *testing.T) {
 	}
 	var sessions []table.Session
 	for i := 0; i < 1000; i++ {
-		err := do(
-			context.Background(),
-			p,
-			config.New(),
+		err := retryBackoff(context.Background(), p,
 			func(ctx context.Context, s table.Session) error {
 				sessions = append(sessions, s)
 				s.(*session).SetStatus(table.SessionClosing)
 
 				return nil
 			},
-			nil,
 		)
 		if err != nil {
 			t.Errorf("unexpected error: %v", err)
@@ -208,14 +200,10 @@ func TestRetryerImmediateReturn(t *testing.T) {
 			p := SingleSession(
 				simpleSession(t),
 			)
-			err := do(
-				context.Background(),
-				p,
-				config.New(),
+			err := retryBackoff(context.Background(), p,
 				func(ctx context.Context, _ table.Session) error {
 					return testErr
 				},
-				nil,
 				retry.WithFastBackoff(
 					testutil.BackoffFunc(func(n int) <-chan time.Time {
 						panic("this code will not be called")
@@ -341,10 +329,7 @@ func TestRetryContextDeadline(t *testing.T) {
 			t.Run(fmt.Sprintf("Timeout=%v,Sleep=%v", timeout, sleep), func(t *testing.T) {
 				ctx, cancel := xcontext.WithTimeout(context.Background(), timeout)
 				defer cancel()
-				_ = do(
-					ctx,
-					p,
-					config.New(),
+				_ = retryBackoff(ctx, p,
 					func(ctx context.Context, _ table.Session) error {
 						select {
 						case <-ctx.Done():
@@ -353,7 +338,6 @@ func TestRetryContextDeadline(t *testing.T) {
 							return errs[r.Int(len(errs))]
 						}
 					},
-					nil,
 				)
 			})
 		}
@@ -442,10 +426,7 @@ func TestRetryWithCustomErrors(t *testing.T) {
 				i        = 0
 				sessions = make(map[table.Session]int)
 			)
-			err := do(
-				ctx,
-				p,
-				config.New(),
+			err := retryBackoff(ctx, p,
 				func(ctx context.Context, s table.Session) (err error) {
 					sessions[s]++
 					i++
@@ -455,7 +436,6 @@ func TestRetryWithCustomErrors(t *testing.T) {
 
 					return nil
 				},
-				nil,
 			)
 			//nolint:nestif
 			if test.retriable {

internal/table/transaction.go

@@ -22,6 +22,7 @@ import (
 var (
 	errTxAlreadyCommitted = xerrors.Wrap(fmt.Errorf("transaction already committed"))
 	errTxRollbackedEarly  = xerrors.Wrap(fmt.Errorf("transaction rollbacked early"))
+	errTxFailedEarly      = xerrors.Wrap(fmt.Errorf("transaction failed early"))
 )
 
 type txState struct {
@@ -42,6 +43,7 @@ const (
 	txStateInitialized txStateEnum = iota
 	txStateCommitted
 	txStateRollbacked
+	txStateFailed
 )
 
 type transaction struct {
@@ -73,11 +75,15 @@ func (tx *transaction) Execute(
 	switch tx.state.Load() {
 	case txStateCommitted:
 		return nil, xerrors.WithStackTrace(errTxAlreadyCommitted)
+	case txStateFailed:
+		return nil, xerrors.WithStackTrace(errTxFailedEarly)
 	case txStateRollbacked:
 		return nil, xerrors.WithStackTrace(errTxRollbackedEarly)
 	default:
 		_, r, err = tx.s.Execute(ctx, tx.control, query, parameters, opts...)
 		if err != nil {
+			tx.state.Store(txStateFailed)
+
 			return nil, xerrors.WithStackTrace(err)
 		}
 
@@ -115,11 +121,15 @@ func (tx *transaction) ExecuteStatement(
 	switch tx.state.Load() {
 	case txStateCommitted:
 		return nil, xerrors.WithStackTrace(errTxAlreadyCommitted)
+	case txStateFailed:
+		return nil, xerrors.WithStackTrace(errTxFailedEarly)
 	case txStateRollbacked:
 		return nil, xerrors.WithStackTrace(errTxRollbackedEarly)
 	default:
 		_, r, err = stmt.Execute(ctx, tx.control, parameters, opts...)
 		if err != nil {
+			tx.state.Store(txStateFailed)
+
 			return nil, xerrors.WithStackTrace(err)
 		}
 
@@ -148,6 +158,8 @@ func (tx *transaction) CommitTx(
 	switch tx.state.Load() {
 	case txStateCommitted:
 		return nil, xerrors.WithStackTrace(errTxAlreadyCommitted)
+	case txStateFailed:
+		return nil, xerrors.WithStackTrace(errTxFailedEarly)
 	case txStateRollbacked:
 		return nil, xerrors.WithStackTrace(errTxRollbackedEarly)
 	default:
@@ -174,6 +186,8 @@ func (tx *transaction) CommitTx(
 
 		response, err = tx.s.tableService.CommitTransaction(ctx, request)
 		if err != nil {
+			tx.state.Store(txStateFailed)
+
 			return nil, xerrors.WithStackTrace(err)
 		}
 
@@ -206,6 +220,8 @@ func (tx *transaction) Rollback(ctx context.Context) (err error) {
 	switch tx.state.Load() {
 	case txStateCommitted:
 		return nil // nop for committed tx
+	case txStateFailed:
+		return xerrors.WithStackTrace(errTxFailedEarly)
 	case txStateRollbacked:
 		return xerrors.WithStackTrace(errTxRollbackedEarly)
 	default:
@@ -222,6 +238,8 @@ func (tx *transaction) Rollback(ctx context.Context) (err error) {
 			},
 		)
 		if err != nil {
+			tx.state.Store(txStateFailed)
+
 			return xerrors.WithStackTrace(err)
 		}
 

retry/sql.go

@@ -161,26 +161,21 @@ func DoTx(ctx context.Context, db *sql.DB, op func(context.Context, *sql.Tx) err
 	}
 	err := Retry(ctx, func(ctx context.Context) (finalErr error) {
 		attempts++
+
 		tx, err := db.BeginTx(ctx, options.txOptions)
 		if err != nil {
 			return unwrapErrBadConn(xerrors.WithStackTrace(err))
 		}
 		defer func() {
-			if finalErr == nil {
-				return
-			}
-			errRollback := tx.Rollback()
-			if errRollback == nil {
-				return
+			if finalErr != nil {
+				_ = tx.Rollback()
 			}
-			finalErr = xerrors.NewWithIssues("",
-				xerrors.WithStackTrace(finalErr),
-				xerrors.WithStackTrace(fmt.Errorf("rollback failed: %w", errRollback)),
-			)
 		}()
+
 		if err = op(xcontext.MarkRetryCall(ctx), tx); err != nil {
 			return unwrapErrBadConn(xerrors.WithStackTrace(err))
 		}
+
 		if err = tx.Commit(); err != nil {
 			return unwrapErrBadConn(xerrors.WithStackTrace(err))
 		}