Merge pull request #852 from ydb-platform/time-after

Removed `backoff.Backoff.Wait` interface method for exclude resource leak with bug-provoked usage of `time.After`

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Removed `internal/backoff.Backoff.Wait` interface method for exclude resource leak with bug-provoked usage of `time.After` method
 * Marked as deprecated `retry.WithDoRetryOptions` and `retry.WithDoTxRetryOptions`
 * Added receiving first result set on construct `internal/table/scanner.NewStream()`
 * Added experimental package `metrics` with SDK metrics

internal/backoff/backoff.go

@@ -4,20 +4,11 @@ import (
 	"math"
 	"time"
 
-	"github.com/jonboulle/clockwork"
-
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xrand"
 )
 
 // Backoff is the interface that contains logic of delaying operation retry.
 type Backoff interface {
-	// Wait maps index of the retry to a channel which fulfillment means that
-	// Delay is over.
-	//
-	// Note that retry index begins from 0 and 0-th index means that it is the
-	// first retry attempt after an initial error.
-	Wait(n int) <-chan time.Time
-
 	// Delay returns mapping of i to Delay.
 	Delay(i int) time.Duration
 }
@@ -39,6 +30,8 @@ var (
 	)
 )
 
+var _ Backoff = (*logBackoff)(nil)
+
 // logBackoff contains logarithmic Backoff policy.
 type logBackoff struct {
 	// slotDuration is a size of a single time slot used in Backoff Delay
@@ -59,8 +52,6 @@ type logBackoff struct {
 	// duration D; and R is a random sized part from [0,(D - F)].
 	jitterLimit float64
 
-	clock clockwork.Clock
-
 	// generator of jitter
 	r xrand.Rand
 }
@@ -85,12 +76,6 @@ func WithJitterLimit(jitterLimit float64) option {
 	}
 }
 
-func WithClock(clock clockwork.Clock) option {
-	return func(b *logBackoff) {
-		b.clock = clock
-	}
-}
-
 func WithSeed(seed int64) option {
 	return func(b *logBackoff) {
 		b.r = xrand.New(xrand.WithLock(), xrand.WithSeed(seed))
@@ -99,8 +84,7 @@ func WithSeed(seed int64) option {
 
 func New(opts ...option) logBackoff {
 	b := logBackoff{
-		r:     xrand.New(xrand.WithLock()),
-		clock: clockwork.NewRealClock(),
+		r: xrand.New(xrand.WithLock()),
 	}
 	for _, o := range opts {
 		if o != nil {
@@ -110,11 +94,6 @@ func New(opts ...option) logBackoff {
 	return b
 }
 
-// Wait implements Backoff interface.
-func (b logBackoff) Wait(n int) <-chan time.Time {
-	return b.clock.After(b.Delay(n))
-}
-
 // Delay returns mapping of i to Delay.
 func (b logBackoff) Delay(i int) time.Duration {
 	s := b.slotDuration

internal/repeater/repeater.go

@@ -168,7 +168,6 @@ func (r *repeater) worker(ctx context.Context, tick clockwork.Ticker) {
 		backoff.WithSlotDuration(500*time.Millisecond),
 		backoff.WithCeiling(6),
 		backoff.WithJitterLimit(1),
-		backoff.WithClock(r.clock),
 	)
 
 	// forceIndex defines delay index for force backoff
@@ -178,12 +177,16 @@ func (r *repeater) worker(ctx context.Context, tick clockwork.Ticker) {
 		if forceIndex == 0 {
 			return EventForce
 		}
+
+		force := r.clock.NewTimer(force.Delay(forceIndex))
+		defer force.Stop()
+
 		select {
 		case <-ctx.Done():
 			return EventCancel
 		case <-tick.Chan():
 			return EventTick
-		case <-force.Wait(forceIndex):
+		case <-force.Chan():
 			return EventForce
 		}
 	}

internal/topic/topicreaderinternal/stream_reconnector.go

@@ -94,11 +94,18 @@ func (r *readerReconnector) ReadMessageBatch(ctx context.Context, opts ReadMessa
 
 	for {
 		if attempt > 0 {
-			select {
-			case <-ctx.Done():
-				return nil, ctx.Err()
-			case <-backoff.Fast.Wait(attempt):
-				// pass
+			if err := func() error {
+				t := r.clock.NewTimer(backoff.Fast.Delay(attempt))
+				defer t.Stop()
+
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case <-t.Chan():
+					return nil
+				}
+			}(); err != nil {
+				return nil, err
 			}
 		}
 
@@ -211,13 +218,18 @@ func (r *readerReconnector) reconnectionLoop(ctx context.Context) {
 				request.reason,
 				r.clock.Since(retriesStarted),
 			); isRetriableErr {
-				delay := retryBackoff.Delay(attempt)
-
-				select {
-				case <-ctx.Done():
+				if err := func() error {
+					t := r.clock.NewTimer(retryBackoff.Delay(attempt))
+					defer t.Stop()
+
+					select {
+					case <-ctx.Done():
+						return ctx.Err()
+					case <-t.Chan():
+						return nil
+					}
+				}(); err != nil {
 					return
-				case <-r.clock.After(delay):
-					// pass
 				}
 			}
 		}

internal/wait/wait.go

@@ -2,6 +2,7 @@ package wait
 
 import (
 	"context"
+	"time"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/backoff"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
@@ -11,8 +12,11 @@ import (
 // expiration.
 // It returns non-nil error if and only if deadline expiration branch wins.
 func waitBackoff(ctx context.Context, b backoff.Backoff, i int) error {
+	t := time.NewTimer(b.Delay(i))
+	defer t.Stop()
+
 	select {
-	case <-b.Wait(i):
+	case <-t.C:
 		return nil
 	case <-ctx.Done():
 		if err := ctx.Err(); err != nil {

internal/xsql/errors.go

@@ -20,7 +20,7 @@ type ErrConnAlreadyHaveTx struct {
 }
 
 func (err *ErrConnAlreadyHaveTx) Error() string {
-	return "conn already have an opened currentTx: " + err.currentTx
+	return "conn already have an open currentTx: " + err.currentTx
 }
 
 func (err *ErrConnAlreadyHaveTx) As(target interface{}) bool {