scripting service client

Author: asmyasnikov

CHANGELOG.md

@@ -1,10 +1,10 @@
 ## 3.7.0
-* Replaced `Option` to `CustomOption` on `Connection` interface methods 
+* Replaced `Option` to `CustomOption` on `Connection` interface methods
 * Implements `WithCustom[Token,Database]` options for redefine database and token
 * Removed experimental `balancer.PreferEndpoints[WithFallback][RegEx]` balancers
-* Implements `balancer.PreferLocations[WithFallback]` balancers
 * Supported connections `TTL` with `Option` `WithConnectionTTL`
-* Remove unnecessary `WithFastDial` option (lazy connections always fast inserts into cluster)
+* Remove unnecessary `WithFastDial` option (lazy connections are always fast inserts into cluster)
+* Added `ScriptingYQL` service client
 
 ## 3.6.2
 * Refactored table retry helpers

connection.go

@@ -20,46 +20,46 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/log"
 	"github.com/ydb-platform/ydb-go-sdk/v3/ratelimiter"
 	"github.com/ydb-platform/ydb-go-sdk/v3/scheme"
+	"github.com/ydb-platform/ydb-go-sdk/v3/scripting"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table"
 	tableConfig "github.com/ydb-platform/ydb-go-sdk/v3/table/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/trace"
 )
 
+// Connection interface provide access to YDB service clients
+// Interface and list of clients may be changed in the future
 type Connection interface {
 	db.Connection
 
 	// Table returns table client with options from Connection instance.
 	// Options provide options replacement for requested table client
-	// such as endpoint, database, secure connection flag and credentials
-	// Options replacement feature not implements now
+	// such as database and access token
 	Table(opts ...CustomOption) table.Client
 
 	// Scheme returns scheme client with options from Connection instance.
 	// Options provide options replacement for requested scheme client
-	// such as endpoint, database, secure connection flag and credentials
-	// Options replacement feature not implements now
+	// such as database and access token
 	Scheme(opts ...CustomOption) scheme.Client
 
 	// Coordination returns coordination client with options from Connection instance.
 	// Options provide options replacement for requested coordination client
-	// such as endpoint, database, secure connection flag and credentials
-	// Options replacement feature not implements now
+	// such as database and access token
 	Coordination(opts ...CustomOption) coordination.Client
 
 	// Ratelimiter returns rate limiter client with options from Connection instance.
 	// Options provide options replacement for requested rate limiter client
-	// such as endpoint, database, secure connection flag and credentials
-	// Options replacement feature not implements now
+	// such as database and access token
 	Ratelimiter(opts ...CustomOption) ratelimiter.Client
 
 	// Discovery returns discovery client with options from Connection instance.
 	// Options provide options replacement for requested discovery client
-	// such as endpoint, database, secure connection flag and credentials
-	// Options replacement feature not implements now
+	// such as database and access token
 	Discovery(opts ...CustomOption) discovery.Client
 
-	// Close clears resources and close all connections to YDB
-	Close(ctx context.Context) error
+	// Scripting returns scripting client with options from Connection instance.
+	// Options provide options replacement for requested discovery client
+	// such as database and access token
+	Scripting(opts ...CustomOption) scripting.Client
 }
 
 type connection struct {
@@ -71,6 +71,7 @@ type connection struct {
 	db           db.Connection
 	table        table.Client
 	scheme       scheme.Client
+	scripting    scripting.Client
 	discovery    discovery.Client
 	coordination coordination.Client
 	rateLimiter  ratelimiter.Client
@@ -95,9 +96,15 @@ func (c *connection) Close(ctx context.Context) error {
 	if err := c.table.Close(ctx); err != nil {
 		issues = append(issues, err)
 	}
+	if err := c.scripting.Close(ctx); err != nil {
+		issues = append(issues, err)
+	}
 	if err := c.db.Close(ctx); err != nil {
 		issues = append(issues, err)
 	}
+	if err := c.conns.Close(ctx); err != nil {
+		issues = append(issues, err)
+	}
 	if len(issues) > 0 {
 		return errors.NewWithIssues("close failed", issues...)
 	}
@@ -170,6 +177,13 @@ func (c *connection) Discovery(opts ...CustomOption) discovery.Client {
 	return proxy.Discovery(c.discovery, c.meta(opts...))
 }
 
+func (c *connection) Scripting(opts ...CustomOption) scripting.Client {
+	if len(opts) == 0 {
+		return c.scripting
+	}
+	return proxy.Scripting(c.scripting, c.meta(opts...))
+}
+
 func (c *connection) meta(opts ...CustomOption) meta.Meta {
 	if len(opts) == 0 {
 		return c.config.Meta()
@@ -225,6 +239,7 @@ func New(ctx context.Context, opts ...Option) (_ Connection, err error) {
 	}
 	c.table = lazy.Table(c.db, c.tableOptions)
 	c.scheme = lazy.Scheme(c.db)
+	c.scripting = lazy.Scripting(c.db)
 	c.discovery = lazy.Discovery(c.db, c.config.Trace())
 	c.coordination = lazy.Coordination(c.db)
 	c.rateLimiter = lazy.Ratelimiter(c.db)

internal/cluster/icluster/cluster.go

@@ -1,9 +1,9 @@
 package icluster
 
 import (
-	"context"
-
 	"google.golang.org/grpc"
+
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/closer"
 )
 
 type Cluster interface {
@@ -12,7 +12,5 @@ type Cluster interface {
 	// Lazy getting grpc-connection must use for embedded client-side balancing
 	// DB may be put into code-generated client constructor as is.
 	grpc.ClientConnInterface
-
-	// Close clears resources and close all connections to YDB
-	Close(ctx context.Context) error
+	closer.Closer
 }

internal/conn/conn.go

@@ -46,6 +46,8 @@ type conn struct {
 }
 
 func (c *conn) Park(ctx context.Context) error {
+	c.Lock()
+	defer c.Unlock()
 	return c.close(ctx)
 }
 
@@ -155,6 +157,7 @@ func isBroken(raw *grpc.ClientConn) bool {
 	return s == connectivity.Shutdown || s == connectivity.TransientFailure
 }
 
+// conn must be locked
 func (c *conn) close(ctx context.Context) (err error) {
 	if c.cc == nil {
 		return nil

internal/conn/pool.go

@@ -8,13 +8,16 @@ import (
 
 	"google.golang.org/grpc"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/closer"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/endpoint"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/errors"
 )
 
 type Pool interface {
+	closer.Closer
+
 	Get(endpoint endpoint.Endpoint) Conn
 	Pessimize(ctx context.Context, e endpoint.Endpoint) error
-	Close()
 }
 
 type PoolConfig interface {
@@ -51,19 +54,24 @@ func (p *pool) Get(endpoint endpoint.Endpoint) Conn {
 	return cc
 }
 
-func (p *pool) Close() {
+func (p *pool) Close(ctx context.Context) error {
 	close(p.done)
+	var issues []error
+	p.mtx.Lock()
+	defer p.mtx.Unlock()
+	for a, c := range p.conns {
+		if err := c.Close(ctx); err != nil {
+			issues = append(issues, err)
+		}
+		delete(p.conns, a)
+	}
+	if len(issues) == 0 {
+		return nil
+	}
+	return errors.NewWithIssues("connection pool close failed", issues...)
 }
 
 func (p *pool) connCloser(ctx context.Context, interval time.Duration) {
-	defer func() {
-		p.mtx.RLock()
-		for a, c := range p.conns {
-			c.Close(ctx)
-			delete(p.conns, a)
-		}
-		p.mtx.RUnlock()
-	}()
 	for {
 		select {
 		case <-p.done:

internal/errors/database.go

@@ -0,0 +1,11 @@
+package errors
+
+type RetryableError struct {
+	Msg               string
+	BackoffType       BackoffType
+	MustDeleteSession bool
+}
+
+func (e *RetryableError) Error() string {
+	return e.Msg
+}

internal/errors/retryable.go

@@ -16,7 +16,7 @@ type lazyCoordination struct {
 	m      sync.Mutex
 }
 
-func Coordination(db db.Connection) *lazyCoordination {
+func Coordination(db db.Connection) coordination.Client {
 	return &lazyCoordination{
 		db: db,
 	}

internal/lazy/coordiantion.go

@@ -17,7 +17,7 @@ type lazyDiscovery struct {
 	m      sync.Mutex
 }
 
-func Discovery(db db.Connection, trace trace.Driver) *lazyDiscovery {
+func Discovery(db db.Connection, trace trace.Driver) discovery.Client {
 	return &lazyDiscovery{
 		db:     db,
 		trace:  trace,