Fixed CHANGELOG.md + refactored retry.Retry

Author: asmyasnikov

CHANGELOG.md

@@ -1,17 +1,14 @@
 ## 3.10.0
-* Added `trace.Discovery`
-* Added `trace.Scripting` stub (will be implements in the future)
-* Added `trace.Scheme` stub (will be implements in the future)
-* Added `trace.Coordination` stub (will be implements in the future)
-* Added `trace.Ratelimiter` stub (will be implements in the future)
+* Extended `trace.Details` constants for support per-service events
+* Added `trace.Discovery` struct for traces discovery events
+* Added `trace.Ratelimiter`, `trace.Coordination`, `trace.Scripting`, `trace.Scheme` stubs (will be implements in the future)
+* Added `ratelimiter/config`, `coordination/config`, `scripting/config`, `scheme/config`, `discovery/config` packages for specify per-service configs
 * Removed `trace.Driver.OnDiscovery` callback (moved to `trace.Discovery`)
-* Added `discovery/config` package for specify discovery config
-* Added `scheme/config` package for specify scheme config
-* Added `scripting/config` package for specify scripting config
-* Added `coordination/config` package for specify coordination config
-* Added `ratelimiter/config` package for specify ratelimiter config
 * Refactored initialization step (firstly makes discovery client)
-* Refactored dns-balancing logic for grpc-go 1.43
+* Removed `internal/lazy.Discovery` (discovery client always initialized)
+* Fixed `trace.Table` event structs
+* Refactored grpc options for define dns-balancing configuration
+* Refactored `retry.Retry` signature (added `retry.WithID`, `retry.WithTrace` and `retry.WithIdempotent` opt-in args, required param `isIdempotentOperation` removed) 
 
 ## 3.9.4
 * Fixed data race on closing session pool

internal/lazy/coordiantion.go

@@ -28,21 +28,21 @@ func Coordination(db db.Connection, options []config.Option) coordination.Client
 
 func (c *lazyCoordination) CreateNode(ctx context.Context, path string, config coordination.NodeConfig) (err error) {
 	c.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return c.client.CreateNode(ctx, path, config)
 	})
 }
 
 func (c *lazyCoordination) AlterNode(ctx context.Context, path string, config coordination.NodeConfig) (err error) {
 	c.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return c.client.AlterNode(ctx, path, config)
 	})
 }
 
 func (c *lazyCoordination) DropNode(ctx context.Context, path string) (err error) {
 	c.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return c.client.DropNode(ctx, path)
 	})
 }
@@ -56,7 +56,7 @@ func (c *lazyCoordination) DescribeNode(
 	err error,
 ) {
 	c.init()
-	err = retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		entry, config, err = c.client.DescribeNode(ctx, path)
 		return err
 	})

internal/lazy/ratelimiter.go

@@ -43,7 +43,7 @@ func (r *lazyRatelimiter) CreateResource(
 	resource ratelimiter.Resource,
 ) (err error) {
 	r.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return r.client.CreateResource(ctx, coordinationNodePath, resource)
 	})
 }
@@ -54,7 +54,7 @@ func (r *lazyRatelimiter) AlterResource(
 	resource ratelimiter.Resource,
 ) (err error) {
 	r.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return r.client.AlterResource(ctx, coordinationNodePath, resource)
 	})
 }
@@ -65,7 +65,7 @@ func (r *lazyRatelimiter) DropResource(
 	resourcePath string,
 ) (err error) {
 	r.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return r.client.DropResource(ctx, coordinationNodePath, resourcePath)
 	})
 }
@@ -77,7 +77,7 @@ func (r *lazyRatelimiter) ListResource(
 	recursive bool,
 ) (paths []string, err error) {
 	r.init()
-	err = retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		paths, err = r.client.ListResource(ctx, coordinationNodePath, resourcePath, recursive)
 		return err
 	})
@@ -90,7 +90,7 @@ func (r *lazyRatelimiter) DescribeResource(
 	resourcePath string,
 ) (resource *ratelimiter.Resource, err error) {
 	r.init()
-	err = retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		resource, err = r.client.DescribeResource(ctx, coordinationNodePath, resourcePath)
 		return err
 	})
@@ -105,7 +105,7 @@ func (r *lazyRatelimiter) AcquireResource(
 	isUsedAmount bool,
 ) (err error) {
 	r.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return r.client.AcquireResource(ctx, coordinationNodePath, resourcePath, amount, isUsedAmount)
 	})
 }

internal/lazy/scheme.go

@@ -31,7 +31,7 @@ func (s *lazyScheme) ModifyPermissions(
 	opts ...scheme.PermissionsOption,
 ) (err error) {
 	s.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return s.client.ModifyPermissions(ctx, path, opts...)
 	})
 }
@@ -58,32 +58,32 @@ func (s *lazyScheme) init() {
 
 func (s *lazyScheme) DescribePath(ctx context.Context, path string) (e scheme.Entry, err error) {
 	s.init()
-	err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		e, err = s.client.DescribePath(ctx, path)
 		return err
-	})
+	}, retry.WithIdempotent())
 	return e, err
 }
 
 func (s *lazyScheme) MakeDirectory(ctx context.Context, path string) (err error) {
 	s.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return s.client.MakeDirectory(ctx, path)
 	})
 }
 
 func (s *lazyScheme) ListDirectory(ctx context.Context, path string) (d scheme.Directory, err error) {
 	s.init()
-	err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		d, err = s.client.ListDirectory(ctx, path)
 		return err
-	})
+	}, retry.WithIdempotent())
 	return d, err
 }
 
 func (s *lazyScheme) RemoveDirectory(ctx context.Context, path string) (err error) {
 	s.init()
-	return retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	return retry.Retry(ctx, func(ctx context.Context) (err error) {
 		return s.client.RemoveDirectory(ctx, path)
 	})
 }

internal/lazy/scripting.go

@@ -26,7 +26,7 @@ func (s *lazyScripting) Execute(
 	params *table.QueryParameters,
 ) (res result.Result, err error) {
 	s.init()
-	err = retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		res, err = s.client.Execute(ctx, query, params)
 		return err
 	})
@@ -39,7 +39,7 @@ func (s *lazyScripting) Explain(
 	mode scripting.ExplainMode,
 ) (e table.ScriptingYQLExplanation, err error) {
 	s.init()
-	err = retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		e, err = s.client.Explain(ctx, query, mode)
 		return err
 	})
@@ -52,7 +52,7 @@ func (s *lazyScripting) StreamExecute(
 	params *table.QueryParameters,
 ) (res result.StreamResult, err error) {
 	s.init()
-	err = retry.Retry(ctx, false, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		res, err = s.client.StreamExecute(ctx, query, params)
 		return err
 	})

internal/lazy/table.go

@@ -27,10 +27,10 @@ func Table(db db.Connection, options []config.Option) table.Client {
 
 func (t *lazyTable) CreateSession(ctx context.Context) (s table.ClosableSession, err error) {
 	t.init(ctx)
-	err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+	err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		s, err = t.client.CreateSession(ctx)
 		return err
-	})
+	}, retry.WithIdempotent())
 	return s, err
 }
 

log/table.go

@@ -266,7 +266,7 @@ func Table(log Logger, details trace.Details) (t trace.Table) {
 				t.OnSessionQueryStreamExecute = func(
 					info trace.SessionQueryStreamExecuteStartInfo,
 				) func(
-					intermediateInfo trace.SessionQueryStreamExecuteIntermediateInfo,
+					trace.SessionQueryStreamExecuteIntermediateInfo,
 				) func(
 					trace.SessionQueryStreamExecuteDoneInfo,
 				) {

retry/retry.go

@@ -62,8 +62,9 @@ func RetryableError(err error, opts ...retryableErrorOption) error {
 }
 
 type retryOptionsHolder struct {
-	id    string
-	trace trace.Retry
+	id         string
+	trace      trace.Retry
+	idempotent bool
 }
 
 type retryOption func(h *retryOptionsHolder)
@@ -81,13 +82,20 @@ func WithTrace(trace trace.Retry) retryOption {
 	}
 }
 
+// WithIdempotent returns discovery trace option
+func WithIdempotent() retryOption {
+	return func(h *retryOptionsHolder) {
+		h.idempotent = true
+	}
+}
+
 // Retry provide the best effort fo retrying operation
 // Retry implements internal busy loop until one of the following conditions is met:
 // - deadline was canceled or deadlined
 // - retry operation returned nil as error
 // Warning: if deadline without deadline or cancellation func Retry will be worked infinite
-// If you need to retry your op func on some logic errors - you must returns from op func RetryableError()
-func Retry(ctx context.Context, isIdempotentOperation bool, op retryOperation, opts ...retryOption) (err error) {
+// If you need to retry your op func on some logic errors - you must return RetryableError() from retryOperation
+func Retry(ctx context.Context, op retryOperation, opts ...retryOption) (err error) {
 	h := &retryOptionsHolder{
 		trace: trace.ContextRetry(ctx),
 	}
@@ -100,7 +108,7 @@ func Retry(ctx context.Context, isIdempotentOperation bool, op retryOperation, o
 		attempts int
 
 		code           = int32(0)
-		onIntermediate = trace.RetryOnRetry(h.trace, ctx, h.id)
+		onIntermediate = trace.RetryOnRetry(h.trace, ctx, h.id, h.idempotent)
 		onDone         func(attempts int, _ error)
 	)
 	defer func() {
@@ -126,7 +134,7 @@ func Retry(ctx context.Context, isIdempotentOperation bool, op retryOperation, o
 			if m.StatusCode() != code {
 				i = 0
 			}
-			if !m.MustRetry(isIdempotentOperation) {
+			if !m.MustRetry(h.idempotent) {
 				return
 			}
 			if e := Wait(ctx, FastBackoff, SlowBackoff, m, i); e != nil {

sugar/sugar.go

@@ -69,10 +69,10 @@ func RemoveRecursive(ctx context.Context, db ydb.Connection, pathToRemove string
 	list = func(i int, p string) error {
 		var dir scheme.Directory
 		var err error
-		err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+		err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 			dir, err = db.Scheme().ListDirectory(ctx, p)
 			return err
-		})
+		}, retry.WithIdempotent())
 		var opErr *errors.OpError
 		if errors.As(err, &opErr) && opErr.Reason == errors.StatusSchemeError {
 			return nil
@@ -91,9 +91,9 @@ func RemoveRecursive(ctx context.Context, db ydb.Connection, pathToRemove string
 				if err = list(i+1, pt); err != nil {
 					return err
 				}
-				err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+				err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 					return db.Scheme().RemoveDirectory(ctx, pt)
-				})
+				}, retry.WithIdempotent())
 				if err != nil {
 					return err
 				}

test/scripting_test.go

@@ -55,7 +55,7 @@ func TestScripting(t *testing.T) {
 		}
 	}()
 	// Execute
-	if err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+	if err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		res, err := db.Scripting().Execute(
 			ctx,
 			"SELECT 1+1",
@@ -87,11 +87,11 @@ func TestScripting(t *testing.T) {
 			return fmt.Errorf("unexpected sum: %v", sum)
 		}
 		return res.Err()
-	}); err != nil {
+	}, retry.WithIdempotent()); err != nil {
 		t.Fatalf("Execute failed: %v", err)
 	}
 	// StreamExecute
-	if err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+	if err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		res, err := db.Scripting().StreamExecute(
 			ctx,
 			"SELECT 1+1",
@@ -124,11 +124,11 @@ func TestScripting(t *testing.T) {
 			return fmt.Errorf("unexpected sum: %v", sum)
 		}
 		return res.Err()
-	}); err != nil {
+	}, retry.WithIdempotent()); err != nil {
 		t.Fatalf("StreamExecute failed: %v", err)
 	}
 	// ExplainPlan
-	if err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+	if err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		res, err := db.Scripting().Explain(
 			ctx,
 			"SELECT 1+1",
@@ -141,11 +141,11 @@ func TestScripting(t *testing.T) {
 			return fmt.Errorf("empty plan")
 		}
 		return nil
-	}); err != nil {
+	}, retry.WithIdempotent()); err != nil {
 		t.Fatalf("Explain failed: %v", err)
 	}
 	// ExplainValidate
-	if err = retry.Retry(ctx, true, func(ctx context.Context) (err error) {
+	if err = retry.Retry(ctx, func(ctx context.Context) (err error) {
 		res, err := db.Scripting().Explain(
 			ctx,
 			"SELECT 1+1",
@@ -158,7 +158,7 @@ func TestScripting(t *testing.T) {
 			return fmt.Errorf("unexpected parameter types")
 		}
 		return nil
-	}); err != nil {
+	}, retry.WithIdempotent()); err != nil {
 		t.Fatalf("Explain failed: %v", err)
 	}
 }