Use fake clock for pauses in TestWriterImpl_Reconnect/ReconnectOnErrors test

Author: rekby

internal/topic/topicwriterinternal/writer_options.go

@@ -3,6 +3,8 @@ package topicwriterinternal
 import (
 	"time"
 
+	"github.com/jonboulle/clockwork"
+
 	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopiccommon"
@@ -152,3 +154,10 @@ func WithTopic(topic string) PublicWriterOption {
 		cfg.topic = topic
 	}
 }
+
+// WithClock is private option for tests
+func WithClock(clock clockwork.Clock) PublicWriterOption {
+	return func(cfg *WriterReconnectorConfig) {
+		cfg.clock = clock
+	}
+}

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -118,7 +118,6 @@ type WriterReconnector struct {
 	queue                          messageQueue
 	background                     background.Worker
 	retrySettings                  topic.RetrySettings
-	clock                          clockwork.Clock
 	writerInstanceID               string
 	sessionID                      string
 	semaphore                      *semaphore.Weighted
@@ -149,7 +148,6 @@ func newWriterReconnectorStopped(
 		cfg:                            cfg,
 		semaphore:                      semaphore.NewWeighted(int64(cfg.MaxQueueLen)),
 		queue:                          newMessageQueue(),
-		clock:                          clockwork.NewRealClock(),
 		lastSeqNo:                      -1,
 		firstInitResponseProcessedChan: make(empty.Chan),
 		encodersMap:                    NewEncoderMap(),
@@ -189,7 +187,7 @@ func (w *WriterReconnector) fillFields(messages []messageWithDataContent) error
 		if w.cfg.AutoSetCreatedTime {
 			if msg.CreatedAt.IsZero() {
 				if now.IsZero() {
-					now = w.clock.Now()
+					now = w.cfg.clock.Now()
 				}
 				msg.CreatedAt = now
 			} else {
@@ -391,17 +389,17 @@ func (w *WriterReconnector) connectionLoop(ctx context.Context) {
 		now := time.Now()
 		if startOfRetries.IsZero() || topic.CheckResetReconnectionCounters(prevAttemptTime, now, w.cfg.connectTimeout) {
 			attempt = 0
-			startOfRetries = w.clock.Now()
+			startOfRetries = w.cfg.clock.Now()
 		} else {
 			attempt++
 		}
 		prevAttemptTime = now
 
 		if reconnectReason != nil {
-			retryDuration := w.clock.Since(startOfRetries)
+			retryDuration := w.cfg.clock.Since(startOfRetries)
 			if backoff, retry := topic.CheckRetryMode(reconnectReason, w.retrySettings, retryDuration); retry {
 				delay := backoff.Delay(attempt)
-				delayTimer := w.clock.NewTimer(delay)
+				delayTimer := w.cfg.clock.NewTimer(delay)
 				select {
 				case <-doneCtx:
 					delayTimer.Stop()

internal/topic/topicwriterinternal/writer_reconnector_test.go

@@ -7,12 +7,14 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"sort"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"github.com/jonboulle/clockwork"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/mock/gomock"
 
@@ -426,7 +428,19 @@ func TestWriterImpl_Reconnect(t *testing.T) {
 	xtest.TestManyTimesWithName(t, "ReconnectOnErrors", func(t testing.TB) {
 		ctx := xtest.Context(t)
 
-		w := newTestWriterStopped()
+		clock := clockwork.NewFakeClock()
+
+		go func() {
+			for {
+				if ctx.Err() != nil {
+					return
+				}
+				clock.Advance(time.Second)
+				time.Sleep(time.Microsecond)
+			}
+		}()
+
+		w := newTestWriterStopped(WithClock(clock), WithTokenUpdateInterval(time.Duration(math.MaxInt64)))
 
 		mc := gomock.NewController(t)