Merge pull request #1448 from ydb-platform/fix-flap-TestDoCreateSessionError

added lastErr from previous attemp in retry.RetryWithResult

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Added `lastErr` from previous attempt in `retry.RetryWithResult`
+
 ## v3.80.0
 * Replaced internal table client pool entities to `internal/pool`
 

internal/credentials/oauth2_test.go

@@ -145,7 +145,7 @@ func TestOauth2TokenExchange(t *testing.T) {
 			Status:            http.StatusForbidden,
 			ExpectedToken:     "",
 			ExpectedError:     errCouldNotExchangeToken,
-			ExpectedErrorPart: "403 Forbidden, description: \"something went bad\", error_uri: my_error_uri",
+			ExpectedErrorPart: "403 Forbidden, description: \\\"something went bad\\\", error_uri: my_error_uri",
 		},
 		{
 			Response:      `{"access_token":"test_token","token_type":"","expires_in":42,"some_other_field":"x"}`,
@@ -170,7 +170,7 @@ func TestOauth2TokenExchange(t *testing.T) {
 			Status:            http.StatusOK,
 			ExpectedToken:     "",
 			ExpectedError:     errDifferentScope,
-			ExpectedErrorPart: "Expected \"test_scope1 test_scope2\", but got \"s\"",
+			ExpectedErrorPart: "Expected \\\"test_scope1 test_scope2\\\", but got \\\"s\\\"",
 		},
 		{
 			Response:      `{"access_token":"","token_type":"Bearer","expires_in":42}`,

internal/pool/pool.go

@@ -348,6 +348,7 @@ func (p *Pool[PT, T]) With(
 	opts ...retry.Option,
 ) (finalErr error) {
 	var attempts int
+
 	if onWith := p.config.trace.OnWith; onWith != nil {
 		onDone := onWith(&ctx,
 			stack.FunctionID("github.com/ydb-platform/ydb-go-sdk/v3/internal/pool.(*Pool).With"),
@@ -608,25 +609,27 @@ func (p *Pool[PT, T]) getItem(ctx context.Context) (item PT, finalErr error) { /
 			}
 		}
 
-		item, createItemErr := p.createItem(ctx)
+		item, err := p.createItem(ctx)
 		if item != nil {
 			return item, nil
 		}
 
-		if !isRetriable(createItemErr) {
-			return nil, xerrors.WithStackTrace(createItemErr)
+		if !isRetriable(err) {
+			return nil, xerrors.WithStackTrace(xerrors.Join(err, lastErr))
 		}
 
-		item, waitFromChErr := p.waitFromCh(ctx)
+		lastErr = err
+
+		item, err = p.waitFromCh(ctx)
 		if item != nil {
 			return item, nil
 		}
 
-		if waitFromChErr != nil && !isRetriable(waitFromChErr) {
-			return nil, xerrors.WithStackTrace(waitFromChErr)
+		if err != nil && !isRetriable(err) {
+			return nil, xerrors.WithStackTrace(xerrors.Join(err, lastErr))
 		}
 
-		lastErr = xerrors.WithStackTrace(xerrors.Join(createItemErr, waitFromChErr))
+		lastErr = err
 	}
 
 	p.mu.RLock()

internal/table/retry_test.go

@@ -132,26 +132,29 @@ func TestDoBadSession(t *testing.T) {
 }
 
 func TestDoCreateSessionError(t *testing.T) {
-	ctx, cancel := xcontext.WithTimeout(xtest.Context(t), time.Second)
-	defer cancel()
-	p := pool.New[*session, session](ctx,
-		pool.WithCreateItemFunc[*session, session](func(ctx context.Context) (*session, error) {
-			return nil, xerrors.Operation(xerrors.WithStatusCode(Ydb.StatusIds_UNAVAILABLE))
-		}),
-		pool.WithSyncCloseItem[*session, session](),
-	)
-	err := do(ctx, p, config.New(),
-		func(ctx context.Context, s table.Session) error {
-			return nil
-		},
-		nil,
-	)
-	if !xerrors.Is(err, context.DeadlineExceeded) {
-		t.Errorf("unexpected error: %v", err)
-	}
-	if !xerrors.IsOperationError(err, Ydb.StatusIds_UNAVAILABLE) {
-		t.Errorf("unexpected error: %v", err)
-	}
+	rootCtx := xtest.Context(t)
+	xtest.TestManyTimes(t, func(t testing.TB) {
+		ctx, cancel := xcontext.WithTimeout(rootCtx, 30*time.Millisecond)
+		defer cancel()
+		p := pool.New[*session, session](ctx,
+			pool.WithCreateItemFunc[*session, session](func(ctx context.Context) (*session, error) {
+				return nil, xerrors.Operation(xerrors.WithStatusCode(Ydb.StatusIds_UNAVAILABLE))
+			}),
+			pool.WithSyncCloseItem[*session, session](),
+		)
+		err := do(ctx, p, config.New(),
+			func(ctx context.Context, s table.Session) error {
+				return nil
+			},
+			nil,
+		)
+		if !xerrors.Is(err, context.DeadlineExceeded) {
+			t.Errorf("unexpected error: %v", err)
+		}
+		if !xerrors.IsOperationError(err, Ydb.StatusIds_UNAVAILABLE) {
+			t.Errorf("unexpected error: %v", err)
+		}
+	})
 }
 
 func TestDoImmediateReturn(t *testing.T) {

retry/retry.go

@@ -319,6 +319,7 @@ func RetryWithResult[T any](ctx context.Context, //nolint:revive,funlen
 	var (
 		i        int
 		attempts int
+		lastErr  error
 
 		code   = int64(0)
 		onDone = trace.RetryOnRetry(options.trace, &ctx,
@@ -333,9 +334,10 @@ func RetryWithResult[T any](ctx context.Context, //nolint:revive,funlen
 		attempts++
 		select {
 		case <-ctx.Done():
-			return zeroValue, xerrors.WithStackTrace(
+			return zeroValue, xerrors.WithStackTrace(xerrors.Join(
 				fmt.Errorf("retry failed on attempt No.%d: %w", attempts, ctx.Err()),
-			)
+				lastErr,
+			))
 
 		default:
 			v, err := opWithRecover(ctx, options, op)
@@ -353,10 +355,11 @@ func RetryWithResult[T any](ctx context.Context, //nolint:revive,funlen
 			code = m.StatusCode()
 
 			if !m.MustRetry(options.idempotent) {
-				return zeroValue, xerrors.WithStackTrace(
+				return zeroValue, xerrors.WithStackTrace(xerrors.Join(
 					fmt.Errorf("non-retryable error occurred on attempt No.%d (idempotent=%v): %w",
 						attempts, options.idempotent, err),
-				)
+					lastErr,
+				))
 			}
 
 			t := time.NewTimer(backoff.Delay(m.BackoffType(), i,
@@ -372,6 +375,7 @@ func RetryWithResult[T any](ctx context.Context, //nolint:revive,funlen
 					xerrors.Join(
 						fmt.Errorf("attempt No.%d: %w", attempts, ctx.Err()),
 						err,
+						lastErr,
 					),
 				)
 			case <-t.C:
@@ -383,10 +387,13 @@ func RetryWithResult[T any](ctx context.Context, //nolint:revive,funlen
 							fmt.Errorf("attempt No.%d: %w", attempts, budget.ErrNoQuota),
 							acquireErr,
 							err,
+							lastErr,
 						),
 					)
 				}
 			}
+
+			lastErr = err
 		}
 	}
 }