Merge branch 'master' into master

Author: korovindenis

.golangci.yml

@@ -218,7 +218,6 @@ linters-settings:
 linters:
   enable-all: true
   disable:
-    - containedctx
     - contextcheck
     - cyclop
     - depguard

CHANGELOG.md

@@ -1,4 +1,8 @@
 * Added type assertion checks to enhance type safety and prevent unexpected panics in critical sections of the codebase
+
+## v3.66.0
+* Added experimental package `retry/budget` for limit second and subsequent retry attempts 
+* Refactored internals for enabling `containedctx` linter
 * Fixed the hanging semaphore issue on coordination session reconnect
 
 ## v3.65.3

config/config.go

@@ -12,6 +12,7 @@ import (
 	balancerConfig "github.com/ydb-platform/ydb-go-sdk/v3/internal/balancer/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/meta"
+	"github.com/ydb-platform/ydb-go-sdk/v3/retry/budget"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
@@ -157,6 +158,13 @@ func WithTrace(t trace.Driver, opts ...trace.DriverComposeOption) Option { //nol
 	}
 }
 
+// Experimental: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#experimental
+func WithRetryBudget(b budget.Budget) Option {
+	return func(c *Config) {
+		config.SetRetryBudget(&c.Common, b)
+	}
+}
+
 func WithTraceRetry(t *trace.Retry, opts ...trace.RetryComposeOption) Option {
 	return func(c *Config) {
 		config.SetTraceRetry(&c.Common, t, opts...)

driver.go

@@ -51,7 +51,6 @@ var _ Connection = (*Driver)(nil)
 
 // Driver type provide access to YDB service clients
 type Driver struct {
-	ctx       context.Context // cancel while Driver.Close called.
 	ctxCancel context.CancelFunc
 
 	userInfo *dsn.UserInfo
@@ -311,7 +310,6 @@ func newConnectionFromOptions(ctx context.Context, opts ...Option) (_ *Driver, e
 
 	d := &Driver{
 		children:  make(map[uint64]*Driver),
-		ctx:       ctx,
 		ctxCancel: driverCtxCancel,
 	}
 

examples/topic/topicreader/topicreader_trace.go

@@ -40,7 +40,7 @@ func ExplicitPartitionStartStopHandler(ctx context.Context, db *ydb.Driver) {
 				) func(
 					trace.TopicReaderPartitionReadStartResponseDoneInfo,
 				) {
-					err := externalSystemLock(info.PartitionContext, info.Topic, info.PartitionID)
+					err := externalSystemLock(*info.PartitionContext, info.Topic, info.PartitionID)
 					if err != nil {
 						stopReader()
 					}
@@ -105,7 +105,7 @@ func PartitionStartStopHandlerAndOwnReadProgressStorage(ctx context.Context, db
 	) func(
 		trace.TopicReaderPartitionReadStartResponseDoneInfo,
 	) {
-		err := externalSystemLock(info.PartitionContext, info.Topic, info.PartitionID)
+		err := externalSystemLock(*info.PartitionContext, info.Topic, info.PartitionID)
 		if err != nil {
 			stopReader()
 		}

internal/background/worker.go

@@ -19,7 +19,7 @@ var (
 
 // A Worker must not be copied after first use
 type Worker struct {
-	ctx            context.Context
+	ctx            context.Context //nolint:containedctx
 	workers        sync.WaitGroup
 	closeReason    error
 	tasksCompleted empty.Chan

internal/backoff/delay.go

@@ -0,0 +1,43 @@
+package backoff
+
+import (
+	"time"
+)
+
+type (
+	delayOptions struct {
+		fast Backoff
+		slow Backoff
+	}
+	delayOption func(o *delayOptions)
+)
+
+func WithFastBackoff(fast Backoff) delayOption {
+	return func(o *delayOptions) {
+		o.fast = fast
+	}
+}
+
+func WithSlowBackoff(slow Backoff) delayOption {
+	return func(o *delayOptions) {
+		o.slow = slow
+	}
+}
+
+func Delay(t Type, i int, opts ...delayOption) time.Duration {
+	optionsHolder := delayOptions{
+		fast: Fast,
+		slow: Slow,
+	}
+	for _, opt := range opts {
+		opt(&optionsHolder)
+	}
+	switch t {
+	case TypeFast:
+		return optionsHolder.fast.Delay(i)
+	case TypeSlow:
+		return optionsHolder.slow.Delay(i)
+	default:
+		return 0
+	}
+}

internal/backoff/delay_test.go

@@ -0,0 +1,92 @@
+package backoff
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
+)
+
+func TestDelay(t *testing.T) {
+	for _, tt := range []struct {
+		name string
+		act  time.Duration
+		exp  time.Duration
+	}{
+		{
+			name: xtest.CurrentFileLine(),
+			act:  Delay(TypeNoBackoff, 0),
+			exp:  0,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act:  Delay(TypeNoBackoff, 1),
+			exp:  0,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act:  Delay(TypeNoBackoff, 2),
+			exp:  0,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act: Delay(TypeFast, 0, WithFastBackoff(New(
+				WithSlotDuration(fastSlot),
+				WithCeiling(6),
+				WithJitterLimit(1),
+			))),
+			exp: 5 * time.Millisecond,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act: Delay(TypeFast, 1, WithFastBackoff(New(
+				WithSlotDuration(fastSlot),
+				WithCeiling(6),
+				WithJitterLimit(1),
+			))),
+			exp: 10 * time.Millisecond,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act: Delay(TypeFast, 3, WithFastBackoff(New(
+				WithSlotDuration(fastSlot),
+				WithCeiling(6),
+				WithJitterLimit(1),
+			))),
+			exp: 40 * time.Millisecond,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act: Delay(TypeSlow, 0, WithSlowBackoff(New(
+				WithSlotDuration(slowSlot),
+				WithCeiling(6),
+				WithJitterLimit(1),
+			))),
+			exp: time.Second,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act: Delay(TypeSlow, 1, WithSlowBackoff(New(
+				WithSlotDuration(slowSlot),
+				WithCeiling(6),
+				WithJitterLimit(1),
+			))),
+			exp: 2 * time.Second,
+		},
+		{
+			name: xtest.CurrentFileLine(),
+			act: Delay(TypeSlow, 3, WithSlowBackoff(New(
+				WithSlotDuration(slowSlot),
+				WithCeiling(6),
+				WithJitterLimit(1),
+			))),
+			exp: 8 * time.Second,
+		},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.exp, tt.act)
+		})
+	}
+}

internal/balancer/balancer.go

@@ -89,6 +89,7 @@ func (b *Balancer) clusterDiscovery(ctx context.Context) (err error) {
 		},
 		retry.WithIdempotent(true),
 		retry.WithTrace(b.driverConfig.TraceRetry()),
+		retry.WithBudget(b.driverConfig.RetryBudget()),
 	)
 }
 

internal/config/config.go

@@ -3,14 +3,18 @@ package config
 import (
 	"time"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/retry/budget"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
+var defaultRetryBudget = budget.Limited(-1)
+
 type Common struct {
 	operationTimeout     time.Duration
 	operationCancelAfter time.Duration
 	disableAutoRetry     bool
 	traceRetry           trace.Retry
+	retryBudget          budget.Budget
 
 	panicCallback func(e interface{})
 }
@@ -48,6 +52,14 @@ func (c *Common) TraceRetry() *trace.Retry {
 	return &c.traceRetry
 }
 
+func (c *Common) RetryBudget() budget.Budget {
+	if c.retryBudget == nil {
+		return defaultRetryBudget
+	}
+
+	return c.retryBudget
+}
+
 // SetOperationTimeout define the maximum amount of time a YDB server will process
 // an operation. After timeout exceeds YDB will try to cancel operation and
 // regardless of the cancellation appropriate error will be returned to
@@ -81,3 +93,7 @@ func SetAutoRetry(c *Common, autoRetry bool) {
 func SetTraceRetry(c *Common, t *trace.Retry, opts ...trace.RetryComposeOption) {
 	c.traceRetry = *c.traceRetry.Compose(t, opts...)
 }
+
+func SetRetryBudget(c *Common, b budget.Budget) {
+	c.retryBudget = b
+}