Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

* Fixed closing of child driver with shared balancer #1541

Merged
merged 2 commits into from
Nov 5, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,3 +1,5 @@
* Fixed closing of child driver with shared balancer

## v3.89.6
* Refactored `database/sql` driver internals for query-service client support in the future

Expand Down
14 changes: 10 additions & 4 deletions driver.go
Original file line number Diff line number Diff line change
Expand Up @@ -109,6 +109,7 @@ type (
balancerWithMeta struct {
balancer *balancer.Balancer
meta *meta.Meta
close func(ctx context.Context) error
}
)

Expand All @@ -134,6 +135,10 @@ func (b *balancerWithMeta) NewStream(ctx context.Context, desc *grpc.StreamDesc,
return b.balancer.NewStream(metaCtx, desc, method, opts...)
}

func (b *balancerWithMeta) Close(ctx context.Context) error {
return b.close(ctx)
}

// Close closes Driver and clear resources
//
//nolint:nonamedreturns
Expand Down Expand Up @@ -176,7 +181,7 @@ func (d *Driver) Close(ctx context.Context) (finalErr error) {
d.query.Close,
d.topic.Close,
d.discovery.Close,
d.metaBalancer.balancer.Close,
d.metaBalancer.Close,
d.pool.Release,
)

Expand Down Expand Up @@ -278,7 +283,7 @@ func Open(ctx context.Context, dsn string, opts ...Option) (_ *Driver, _ error)
}
}

d, err := newConnectionFromOptions(ctx, opts...)
d, err := driverFromOptions(ctx, opts...)
if err != nil {
return nil, xerrors.WithStackTrace(err)
}
Expand Down Expand Up @@ -320,7 +325,7 @@ func MustOpen(ctx context.Context, dsn string, opts ...Option) *Driver {
// Will be removed after Oct 2024.
// Read about versioning policy: https://github.com/ydb-platform/ydb-go-sdk/blob/master/VERSIONING.md#deprecated
func New(ctx context.Context, opts ...Option) (_ *Driver, err error) { //nolint:nonamedreturns
d, err := newConnectionFromOptions(ctx, opts...)
d, err := driverFromOptions(ctx, opts...)
if err != nil {
return nil, xerrors.WithStackTrace(err)
}
Expand All @@ -342,7 +347,7 @@ func New(ctx context.Context, opts ...Option) (_ *Driver, err error) { //nolint:
}

//nolint:cyclop, nonamedreturns, funlen
func newConnectionFromOptions(ctx context.Context, opts ...Option) (_ *Driver, err error) {
func driverFromOptions(ctx context.Context, opts ...Option) (_ *Driver, err error) {
ctx, driverCtxCancel := xcontext.WithCancel(xcontext.ValueOnly(ctx))
defer func() {
if err != nil {
Expand Down Expand Up @@ -444,6 +449,7 @@ func (d *Driver) connect(ctx context.Context) (err error) {
return xerrors.WithStackTrace(err)
}
d.metaBalancer.balancer = b
d.metaBalancer.close = b.Close
}
d.metaBalancer.meta = d.config.Meta()

Expand Down
2 changes: 1 addition & 1 deletion dsn_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -178,7 +178,7 @@ func TestParse(t *testing.T) {
require.ErrorIs(t, err, tt.err)
} else {
require.NoError(t, err)
d, err := newConnectionFromOptions(context.Background(), opts...)
d, err := driverFromOptions(context.Background(), opts...)
require.NoError(t, err)
exp := newConnector(tt.connectorOpts...)
act := newConnector(d.databaseSQLOptions...)
Expand Down
1 change: 1 addition & 0 deletions options.go
Original file line number Diff line number Diff line change
Expand Up @@ -602,6 +602,7 @@ func WithPanicCallback(panicCallback func(e interface{})) Option {
func WithSharedBalancer(parent *Driver) Option {
return func(ctx context.Context, d *Driver) error {
d.metaBalancer.balancer = parent.metaBalancer.balancer
d.metaBalancer.close = func(ctx context.Context) error { return nil }

return nil
}
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,12 @@ import (
"github.com/ydb-platform/ydb-go-sdk/v3/internal/xtest"
"github.com/ydb-platform/ydb-go-sdk/v3/log"
"github.com/ydb-platform/ydb-go-sdk/v3/retry"
"github.com/ydb-platform/ydb-go-sdk/v3/table/result/indexed"
"github.com/ydb-platform/ydb-go-sdk/v3/trace"
)

//nolint:gocyclo
func TestConnection(sourceTest *testing.T) {
func TestDriver(sourceTest *testing.T) {
t := xtest.MakeSyncedTest(sourceTest)
const sumColumn = "sum"
var (
Expand Down Expand Up @@ -164,6 +165,22 @@ func TestConnection(sourceTest *testing.T) {
t.Fatalf("close failed: %+v", e)
}
}()
t.Run("With", func(t *testing.T) {
t.Run("WithSharedBalancer", func(t *testing.T) {
child, err := db.With(ctx, ydb.WithSharedBalancer(db))
require.NoError(t, err)
result, err := child.Scripting().Execute(ctx, `SELECT 1`, nil)
require.NoError(t, err)
require.NoError(t, result.NextResultSetErr(ctx))
require.True(t, result.NextRow())
var value int32
err = result.Scan(indexed.Required(&value))
require.NoError(t, err)
require.EqualValues(t, 1, value)
err = child.Close(ctx)
require.NoError(t, err)
})
})
t.Run("discovery.WhoAmI", func(t *testing.T) {
if err = retry.Retry(ctx, func(ctx context.Context) (err error) {
discoveryClient := Ydb_Discovery_V1.NewDiscoveryServiceClient(ydb.GRPCConn(db))
Expand Down
2 changes: 1 addition & 1 deletion with.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ var nextID atomic.Uint64 //nolint:gochecknoglobals
func (d *Driver) with(ctx context.Context, opts ...Option) (*Driver, uint64, error) {
id := nextID.Add(1)

child, err := newConnectionFromOptions(
child, err := driverFromOptions(
ctx,
append(
append(
Expand Down
2 changes: 1 addition & 1 deletion with_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ func TestWithCertificatesCached(t *testing.T) {
},
} {
t.Run(test.name, func(t *testing.T) {
db, err := newConnectionFromOptions(ctx,
db, err := driverFromOptions(ctx,
append(
test.options,
withConnPool(conn.NewPool(context.Background(), config.New())), //nolint:contextcheck
Expand Down
Loading