Merge pull request #1006 from art22m/inf-reconnection-fix

Fix topic writer infinite reconnections

Author: rekby

internal/topic/retriable_error.go

@@ -7,12 +7,14 @@ import (
 	"time"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/backoff"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/value"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
 )
 
 const (
-	DefaultStartTimeout = time.Minute
+	DefaultStartTimeout          = time.Minute
+	connectionEstablishedTimeout = time.Minute
 )
 
 type RetrySettings struct {
@@ -44,6 +46,9 @@ var (
 
 func CheckResetReconnectionCounters(lastTry, now time.Time, connectionTimeout time.Duration) bool {
 	const resetAttemptEmpiricalCoefficient = 10
+	if connectionTimeout == value.InfiniteDuration {
+		return now.Sub(lastTry) > connectionEstablishedTimeout
+	}
 	return now.Sub(lastTry) > connectionTimeout*resetAttemptEmpiricalCoefficient
 }
 
@@ -87,9 +92,10 @@ func CheckRetryMode(err error, settings RetrySettings, retriesDuration time.Dura
 		return nil, false
 	}
 
-	if mode.BackoffType() == backoff.TypeFast {
+	switch mode.BackoffType() {
+	case backoff.TypeFast:
 		return backoff.Fast, true
+	default:
+		return backoff.Slow, true
 	}
-
-	return backoff.Slow, true
 }

internal/topic/retriable_error_test.go

@@ -12,6 +12,7 @@ import (
 	grpcStatus "google.golang.org/grpc/status"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/backoff"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/value"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 )
 
@@ -232,3 +233,45 @@ func TestCheckRetryMode(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckResetReconnectionCounters(t *testing.T) {
+	now := time.Now()
+	table := []struct {
+		name              string
+		lastTry           time.Time
+		connectionTimeout time.Duration
+		shouldReset       bool
+	}{
+		{
+			name:              "RecentLastTryWithInfiniteConnectionTimeout",
+			lastTry:           now.Add(-30 * time.Second),
+			connectionTimeout: value.InfiniteDuration,
+			shouldReset:       false,
+		},
+		{
+			name:              "OldLastTryWithInfiniteConnectionTimeout",
+			lastTry:           now.Add(-30 * time.Minute),
+			connectionTimeout: value.InfiniteDuration,
+			shouldReset:       true,
+		},
+		{
+			name:              "LastTryLessThanConnectionTimeout",
+			lastTry:           now.Add(-30 * time.Second),
+			connectionTimeout: time.Minute,
+			shouldReset:       false,
+		},
+		{
+			name:              "LastTryGreaterThanConnectionTimeout",
+			lastTry:           now.Add(-time.Hour),
+			connectionTimeout: time.Minute,
+			shouldReset:       true,
+		},
+	}
+
+	for _, test := range table {
+		t.Run(test.name, func(t *testing.T) {
+			shouldReset := CheckResetReconnectionCounters(test.lastTry, now, test.connectionTimeout)
+			require.Equal(t, test.shouldReset, shouldReset)
+		})
+	}
+}

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -369,7 +369,7 @@ func (w *WriterReconnector) connectionLoop(ctx context.Context) {
 		streamCtx, streamCtxCancel = createStreamContext()
 
 		now := time.Now()
-		if topic.CheckResetReconnectionCounters(prevAttemptTime, now, w.cfg.connectTimeout) {
+		if startOfRetries.IsZero() || topic.CheckResetReconnectionCounters(prevAttemptTime, now, w.cfg.connectTimeout) {
 			attempt = 0
 			startOfRetries = w.clock.Now()
 		} else {

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -524,7 +524,7 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 		err := w.Write(ctx, newTestMessages(1))
 		require.NoError(t, err)
 
-		xtest.WaitChannelClosed(t, connectionLoopStopped)
+		xtest.WaitChannelClosedWithTimeout(t, connectionLoopStopped, 4*time.Second)
 	})
 }