sync

Author: rekby

internal/topic/retriable_error.go

@@ -1,10 +1,17 @@
 package topic
 
 import (
+	"time"
+
 	"github.com/ydb-platform/ydb-go-sdk/v3/retry"
 )
 
 func IsRetryableError(err error) bool {
 	mode := retry.Check(err)
 	return mode.MustRetry(true)
 }
+
+func CheckResetReconnectionCounters(lastTry, now time.Time, connectionTimeout time.Duration) bool {
+	const resetAttemptEmpiricalCoefficient = 10
+	return now.Sub(lastTry) > connectionTimeout*resetAttemptEmpiricalCoefficient
+}

internal/topic/topicreaderinternal/stream_reconnector.go

@@ -169,15 +169,12 @@ func (r *readerReconnector) reconnectionLoop(ctx context.Context) {
 	attempt := 0
 	for {
 		now := r.clock.Now()
-		sinceLastTime := now.Sub(lastTime)
-		lastTime = now
-
-		const resetAttemptEmpiricalCoefficient = 10
-		if sinceLastTime > r.connectTimeout*resetAttemptEmpiricalCoefficient {
+		if topic.CheckResetReconnectionCounters(lastTime, now, r.connectTimeout) {
 			attempt = 0
 		} else {
 			attempt++
 		}
+		lastTime = now
 
 		var request reconnectRequest
 		select {

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -337,6 +337,9 @@ func (w *WriterReconnector) connectionLoop(ctx context.Context) {
 		streamCtxCancel(xerrors.WithStackTrace(errCloseWriterReconnectorConnectionLoop))
 	}()
 
+	var prevConnectionError error
+	var prevConnectionTime time.Time
+
 	for {
 		if ctx.Err() != nil {
 			return