fix inf reconnections

Author: art22m

internal/topic/retriable_error.go

@@ -7,6 +7,7 @@ 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"
 )
@@ -44,6 +45,9 @@ var (
 
 func CheckResetReconnectionCounters(lastTry, now time.Time, connectionTimeout time.Duration) bool {
 	const resetAttemptEmpiricalCoefficient = 10
+	if connectionTimeout == value.InfiniteDuration {
+		return false
+	}
 	return now.Sub(lastTry) > connectionTimeout*resetAttemptEmpiricalCoefficient
 }
 

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,40 @@ func TestCheckRetryMode(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckResetReconnectionCounters(t *testing.T) {
+	now := time.Now()
+	fmt.Println(now.Sub(now.Add(-61 * time.Second)))
+	table := []struct {
+		name              string
+		lastTry           time.Time
+		connectionTimeout time.Duration
+		shouldReset       bool
+	}{
+		{
+			name:              "InfiniteConnectionTimeout",
+			lastTry:           time.Time{},
+			connectionTimeout: value.InfiniteDuration,
+			shouldReset:       false,
+		},
+		{
+			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 {