Merge pull request #1291 Fixed handle unretryable reconnection error in writer reconnection loop.

Author: rekby

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Fixed handle reconnection timeout error
 * Fixed experimental topic listener handle stop partition event
 
 ## v3.76.1

internal/topic/retriable_error.go

@@ -17,6 +17,8 @@ const (
 	connectionEstablishedTimeout = time.Minute
 )
 
+var errNil = xerrors.Wrap(errors.New("nil error is not retrieable"))
+
 type RetrySettings struct {
 	StartTimeout time.Duration // Full retry timeout
 	CheckError   PublicCheckErrorRetryFunction
@@ -53,50 +55,58 @@ func CheckResetReconnectionCounters(lastTry, now time.Time, connectionTimeout ti
 	return now.Sub(lastTry) > connectionTimeout*resetAttemptEmpiricalCoefficient
 }
 
-func CheckRetryMode(err error, settings RetrySettings, retriesDuration time.Duration) (
+// RetryDecision check if err is retriable.
+// if return nil stopRetryReason - err can be retried
+// if return non nil stopRetryReason - err is not retriable and stopRetryReason contains reason,
+// which should be used instead of err
+func RetryDecision(checkErr error, settings RetrySettings, retriesDuration time.Duration) (
 	_ backoff.Backoff,
-	isRetriable bool,
+	stopRetryReason error,
 ) {
 	// nil is not error and doesn't need retry it.
-	if err == nil {
-		return nil, false
+	if checkErr == nil {
+		return nil, xerrors.WithStackTrace(errNil)
 	}
 
 	// eof is retriable for topic
-	if errors.Is(err, io.EOF) && xerrors.RetryableError(err) == nil {
-		err = xerrors.Retryable(err, xerrors.WithName("TopicEOF"))
+	if errors.Is(checkErr, io.EOF) && xerrors.RetryableError(checkErr) == nil {
+		checkErr = xerrors.Retryable(checkErr, xerrors.WithName("TopicEOF"))
 	}
 
 	if retriesDuration > settings.StartTimeout {
-		return nil, false
+		return nil, fmt.Errorf("ydb: topic reader reconnection timeout, last error: %w", xerrors.Unretryable(checkErr))
 	}
 
-	mode := retry.Check(err)
+	mode := retry.Check(checkErr)
 
 	decision := PublicRetryDecisionDefault
 	if settings.CheckError != nil {
-		decision = settings.CheckError(NewCheckRetryArgs(err))
+		decision = settings.CheckError(NewCheckRetryArgs(checkErr))
 	}
 
 	switch decision {
 	case PublicRetryDecisionDefault:
-		isRetriable = mode.MustRetry(true)
+		isRetriable := mode.MustRetry(true)
+		if !isRetriable {
+			return nil, fmt.Errorf("ydb: topic reader unretriable error: %w", xerrors.Unretryable(checkErr))
+		}
 	case PublicRetryDecisionRetry:
-		isRetriable = true
+		// pass
 	case PublicRetryDecisionStop:
-		isRetriable = false
+		return nil, fmt.Errorf(
+			"ydb: topic reader unretriable error by check error callback: %w",
+			xerrors.Unretryable(checkErr),
+		)
 	default:
 		panic(fmt.Errorf("unexpected retry decision: %v", decision))
 	}
 
-	if !isRetriable {
-		return nil, false
-	}
+	// checkErr is retryable error
 
 	switch mode.BackoffType() {
 	case backoff.TypeFast:
-		return backoff.Fast, true
+		return backoff.Fast, nil
 	default:
-		return backoff.Slow, true
+		return backoff.Slow, nil
 	}
 }

internal/topic/retriable_error_test.go

@@ -227,9 +227,9 @@ func TestCheckRetryMode(t *testing.T) {
 
 	for _, test := range table {
 		t.Run(test.name, func(t *testing.T) {
-			resBackoff, retriable := CheckRetryMode(test.err, test.settings, test.duration)
+			resBackoff, stopReason := RetryDecision(test.err, test.settings, test.duration)
 			require.Equal(t, test.resBackoff, resBackoff)
-			require.Equal(t, test.resRetriable, retriable)
+			require.Equal(t, test.resRetriable, stopReason == nil)
 		})
 	}
 }

internal/topic/topicreaderinternal/stream_reconnector.go

@@ -216,11 +216,14 @@ func (r *readerReconnector) reconnectionLoop(ctx context.Context) {
 			}
 		}
 
+		onReconnectionDone := trace.TopicOnReaderReconnect(r.tracer, request.reason)
+
 		if request.reason != nil {
-			if retryBackoff, isRetriableErr := r.checkErrRetryMode(
+			retryBackoff, stopRetryReason := r.checkErrRetryMode(
 				request.reason,
 				r.clock.Since(retriesStarted),
-			); isRetriableErr {
+			)
+			if stopRetryReason == nil {
 				if err := func() error {
 					t := r.clock.NewTimer(retryBackoff.Delay(attempt))
 					defer t.Stop()
@@ -234,10 +237,16 @@ func (r *readerReconnector) reconnectionLoop(ctx context.Context) {
 				}(); err != nil {
 					return
 				}
+			} else {
+				_ = r.CloseWithError(ctx, stopRetryReason)
+				onReconnectionDone(stopRetryReason)
+
+				return
 			}
 		}
 
-		_ = r.reconnect(ctx, request.reason, request.oldReader)
+		err := r.reconnect(ctx, request.reason, request.oldReader)
+		onReconnectionDone(err)
 	}
 }
 
@@ -309,16 +318,16 @@ func (r *readerReconnector) reconnect(ctx context.Context, reason error, oldRead
 }
 
 func (r *readerReconnector) isRetriableError(err error) bool {
-	_, res := topic.CheckRetryMode(err, r.retrySettings, 0)
+	_, stopReason := topic.RetryDecision(err, r.retrySettings, 0)
 
-	return res
+	return stopReason == nil
 }
 
 func (r *readerReconnector) checkErrRetryMode(err error, retriesDuration time.Duration) (
 	backoffType backoff.Backoff,
-	isRetriableErr bool,
+	stopRetryReason error,
 ) {
-	return topic.CheckRetryMode(err, r.retrySettings, retriesDuration)
+	return topic.RetryDecision(err, r.retrySettings, retriesDuration)
 }
 
 func (r *readerReconnector) connectWithTimeout() (_ batchedStreamReader, _ context.CancelCauseFunc, err error) {

internal/topic/topicreaderinternal/stream_reconnector_test.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/background"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/empty"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicreadercommon"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/value"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xcontext"
@@ -65,6 +66,9 @@ func TestTopicReaderReconnectorReadMessageBatch(t *testing.T) {
 
 				return baseReader, nil
 			},
+			retrySettings: topic.RetrySettings{
+				StartTimeout: topic.DefaultStartTimeout,
+			},
 			streamErr: errUnconnected,
 			tracer:    &trace.Topic{},
 		}
@@ -109,6 +113,9 @@ func TestTopicReaderReconnectorReadMessageBatch(t *testing.T) {
 
 				return readers[connectCalled-1], nil
 			},
+			retrySettings: topic.RetrySettings{
+				StartTimeout: topic.DefaultStartTimeout,
+			},
 			streamErr: errUnconnected,
 			tracer:    &trace.Topic{},
 		}
@@ -228,9 +235,12 @@ func TestTopicReaderReconnectorConnectionLoop(t *testing.T) {
 		newStream2.EXPECT().CloseWithError(gomock.Any(), gomock.Any()).MinTimes(1)
 
 		reconnector := &readerReconnector{
-			connectTimeout: value.InfiniteDuration,
-			background:     *background.NewWorker(ctx, "test-worker, "+t.Name()),
+			background: *background.NewWorker(ctx, "test-worker, "+t.Name()),
+			retrySettings: topic.RetrySettings{
+				StartTimeout: value.InfiniteDuration,
+			},
 			tracer:         &trace.Topic{},
+			connectTimeout: value.InfiniteDuration,
 		}
 		reconnector.initChannelsAndClock()
 
@@ -499,7 +509,7 @@ func TestTopicReaderReconnectorReconnectWithError(t *testing.T) {
 		tracer:    &trace.Topic{},
 	}
 	reconnector.initChannelsAndClock()
-	err := reconnector.reconnect(ctx, nil, nil)
+	err := reconnector.reconnect(ctx, errors.New("test-reconnect"), nil)
 	require.ErrorIs(t, err, testErr)
 	require.ErrorIs(t, reconnector.streamErr, testErr)
 }

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -431,7 +431,11 @@ func (w *WriterReconnector) handleReconnectRetry(
 	startOfRetries time.Time,
 ) bool {
 	retryDuration := w.cfg.clock.Since(startOfRetries)
-	if backoff, retry := topic.CheckRetryMode(reconnectReason, w.retrySettings, retryDuration); retry {
+	if backoff, stopRetryReason := topic.RetryDecision(
+		reconnectReason,
+		w.retrySettings,
+		retryDuration,
+	); stopRetryReason == nil {
 		delay := backoff.Delay(attempt)
 		delayTimer := w.cfg.clock.NewTimer(delay)
 		select {
@@ -444,7 +448,7 @@ func (w *WriterReconnector) handleReconnectRetry(
 			// pass
 		}
 	} else {
-		_ = w.close(ctx, fmt.Errorf("%w, was retried (%v)", reconnectReason, retryDuration))
+		_ = w.close(ctx, fmt.Errorf("%w, was retried (%v)", stopRetryReason, retryDuration))
 
 		return true
 	}

internal/xerrors/retryable.go

@@ -123,10 +123,27 @@ func Retryable(err error, opts ...RetryableErrorOption) error {
 
 // RetryableError return Error if err is retriable error, else nil
 func RetryableError(err error) Error {
+	var unretriableErr unretryableError
+	if errors.As(err, &unretriableErr) {
+		return nil
+	}
+
 	var e *retryableError
 	if errors.As(err, &e) {
 		return e
 	}
 
 	return nil
 }
+
+func Unretryable(err error) unretryableError {
+	return unretryableError{err}
+}
+
+type unretryableError struct {
+	error
+}
+
+func (e unretryableError) Unwrap() error {
+	return e.error
+}

internal/xerrors/retryable_test.go

@@ -1,6 +1,7 @@
 package xerrors
 
 import (
+	"errors"
 	"fmt"
 	"testing"
 
@@ -51,3 +52,22 @@ func TestRetryableCode(t *testing.T) {
 		})
 	}
 }
+
+func TestRetriableError(t *testing.T) {
+	t.Run("retryable", func(t *testing.T) {
+		retriable := Retryable(errors.New("test"))
+		wrapped := fmt.Errorf("wrap: %w", retriable)
+		require.Equal(t, retriable, RetryableError(retriable))
+		require.Equal(t, retriable, RetryableError(wrapped))
+	})
+	t.Run("unretryable", func(t *testing.T) {
+		require.NoError(t, RetryableError(errors.New("test")))
+		require.NoError(t, RetryableError(Unretryable(Retryable(errors.New("test")))))
+	})
+}
+
+func TestUnretryableUnwrap(t *testing.T) {
+	test := errors.New("test")
+	wrapped := Unretryable(test)
+	require.ErrorIs(t, wrapped, test)
+}