Merge pull request #344 from ydb-platform/retry-do-tx

* Improved the `xsql` errors mapping to `driver.ErrBadConn`

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,9 @@
+* Improved the `xsql` errors mapping to `driver.ErrBadConn` 
+* Extended `retry.DoTx` test for to achieve equivalence with `retry.Retry` behaviour
+* Added `database/sql` events for tracing `database/sql` driver events
+* Added internal logging for `database/sql` events
+* Supports `YDB_LOG_DETAILS` environment variable for specify scope of log messages
+
 ## v3.33.0
 * Added `retry.DoTx` helper for retrying `database/sql` transactions 
 * Implemented `database/sql` driver over `ydb-go-sdk`

README.md

@@ -126,6 +126,7 @@ Next packages provide debug tooling:
 |----------------------------------|-----------|---------|--------------------------------------------------------------------------------------------------------------------------|
 | `YDB_SSL_ROOT_CERTIFICATES_FILE` | `string`  |         | path to certificates file                                                                                                |
 | `YDB_LOG_SEVERITY_LEVEL`         | `string`  | `quiet` | severity logging level of internal driver logger. Supported: `trace`, `debug`, `info`, `warn`, `error`, `fatal`, `quiet` |
+| `YDB_LOG_DETAILS`                | `string`  | `.*`    | pattern for lookup internal logger logs                                                                                  |
 | `YDB_LOG_NO_COLOR`               | `bool`    | `false` | set any non empty value to disable colouring logs with internal driver logger                                            |
 | `GRPC_GO_LOG_VERBOSITY_LEVEL`    | `integer` |         | set to `99` to see grpc logs                                                                                             |
 | `GRPC_GO_LOG_SEVERITY_LEVEL`     | `string`  |         | set to `info` to see grpc logs                                                                                           |

connection.go

@@ -26,6 +26,7 @@ import (
 	tableConfig "github.com/ydb-platform/ydb-go-sdk/v3/internal/table/config"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicclientinternal"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsql"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xsync"
 	"github.com/ydb-platform/ydb-go-sdk/v3/log"
 	"github.com/ydb-platform/ydb-go-sdk/v3/ratelimiter"
@@ -113,6 +114,8 @@ type connection struct {
 	topic        *topicclientinternal.Client
 	topicOptions []topicoptions.TopicOption
 
+	databaseSQLOptions []xsql.ConnectorOption
+
 	pool *conn.Pool
 
 	mtx      sync.Mutex
@@ -367,7 +370,10 @@ func open(ctx context.Context, opts ...Option) (_ Connection, err error) {
 			opts = append(
 				opts,
 				WithLogger(
-					trace.DetailsAll,
+					trace.MatchDetails(
+						os.Getenv("YDB_LOG_DETAILS"),
+						trace.WithDefaultDetails(trace.DetailsAll),
+					),
 					WithNamespace("ydb"),
 					WithMinLevel(log.FromString(logLevel)),
 					WithNoColor(os.Getenv("YDB_LOG_NO_COLOR") != ""),

internal/retry/check.go

@@ -25,3 +25,26 @@ func Check(err error) (
 		backoff.TypeNoBackoff,
 		false
 }
+
+func MustDeleteSession(err error) bool {
+	var e xerrors.Error
+	if xerrors.As(err, &e) {
+		return e.MustDeleteSession()
+	}
+	return false
+}
+
+func MustRetry(err error, isOperationIdempotent bool) bool {
+	var e xerrors.Error
+	if xerrors.As(err, &e) {
+		switch e.OperationStatus() {
+		case operation.Finished:
+			return false
+		case operation.Undefined:
+			return isOperationIdempotent
+		default:
+			return true
+		}
+	}
+	return false
+}

internal/retry/context.go

@@ -0,0 +1,19 @@
+package retry
+
+import "context"
+
+type ctxIdempotentKey struct{}
+
+func WithIdempotent(ctx context.Context, idempotent bool) context.Context {
+	return context.WithValue(ctx,
+		ctxIdempotentKey{},
+		idempotent,
+	)
+}
+
+func IsIdempotent(ctx context.Context) bool {
+	if idempotent, ok := ctx.Value(ctxIdempotentKey{}).(bool); ok {
+		return idempotent
+	}
+	return false
+}

internal/table/client.go

@@ -81,9 +81,9 @@ type Client struct {
 	index             map[*session]sessionInfo
 	createInProgress  int           // KIKIMR-9163: in-create-process counter
 	limit             int           // Upper bound for Client size.
-	idle              *list.List    // list<table.session>
-	waitq             *list.List    // list<*chan table.session>
-	keeperWake        chan struct{} // Set by internalPoolKeeper.
+	idle              *list.List    // list<*session>
+	waitq             *list.List    // list<*chan *session>
+	keeperWake        chan struct{} // Set by keeper.
 	keeperStop        chan struct{}
 	keeperDone        chan struct{}
 	touchingDone      chan struct{}
@@ -199,32 +199,9 @@ func (c *Client) isClosed() bool {
 
 // c.mu must NOT be held.
 func (c *Client) internalPoolCreateSession(ctx context.Context) (s *session, err error) {
-	defer func() {
-		if s != nil {
-			s.onClose = append(s.onClose, func(s *session) {
-				c.spawnedGoroutines.Start("onClose", func(ctx context.Context) {
-					c.mu.WithLock(func() {
-						info, has := c.index[s]
-						if !has {
-							panic("session removed from pool early")
-						}
-
-						delete(c.index, s)
-
-						trace.TableOnPoolSessionRemove(c.config.Trace(), s)
-						trace.TableOnPoolStateChange(c.config.Trace(), len(c.index), "remove")
-
-						c.internalPoolNotify(nil)
-
-						if info.idle != nil {
-							c.idle.Remove(info.idle)
-						}
-					})
-				})
-			})
-		}
-	}()
-
+	if c.isClosed() {
+		return nil, errClosedClient
+	}
 	// pre-check the Client size
 	var enoughSpace bool
 	c.mu.WithLock(func() {
@@ -272,8 +249,29 @@ func (c *Client) internalPoolCreateSession(ctx context.Context) (s *session, err
 			c.createInProgress--
 			if s != nil {
 				c.index[s] = sessionInfo{}
+
 				trace.TableOnPoolSessionAdd(c.config.Trace(), s)
 				trace.TableOnPoolStateChange(c.config.Trace(), len(c.index), "append")
+
+				s.onClose = append(s.onClose, func(s *session) {
+					c.mu.WithLock(func() {
+						info, has := c.index[s]
+						if !has {
+							return
+						}
+
+						delete(c.index, s)
+
+						trace.TableOnPoolSessionRemove(c.config.Trace(), s)
+						trace.TableOnPoolStateChange(c.config.Trace(), len(c.index), "remove")
+
+						c.internalPoolNotify(nil)
+
+						if info.idle != nil {
+							c.idle.Remove(info.idle)
+						}
+					})
+				})
 			}
 		})
 
@@ -507,6 +505,7 @@ func (c *Client) Close(ctx context.Context) (err error) {
 	var issues []error
 	if atomic.CompareAndSwapUint32(&c.closed, 0, 1) {
 		close(c.done)
+		var toClose []*session
 		c.mu.WithLock(func() {
 			keeperDone := c.keeperDone
 			if ch := c.keeperStop; ch != nil {
@@ -519,13 +518,17 @@ func (c *Client) Close(ctx context.Context) (err error) {
 
 			c.limit = 0
 
-			issues = make([]error, 0, len(c.index))
-			for e := c.idle.Front(); e != nil; e = e.Next() {
-				if err = c.internalPoolCloseSession(ctx, e.Value.(*session)); err != nil {
-					issues = append(issues, err)
-				}
+			toClose = make([]*session, 0, len(c.index))
+			for s := range c.index {
+				toClose = append(toClose, s)
 			}
 		})
+		issues = make([]error, 0, len(toClose))
+		for _, s := range toClose {
+			if err = c.internalPoolCloseSession(ctx, s); err != nil {
+				issues = append(issues, err)
+			}
+		}
 	}
 
 	_ = c.spawnedGoroutines.Close(ctx, errClosedClient)

internal/table/client_test.go

@@ -243,85 +243,54 @@ func TestSessionPoolCloseWhenWaiting(t *testing.T) {
 }
 
 func TestSessionPoolClose(t *testing.T) {
-	wg := sync.WaitGroup{}
-	p := newClientWithStubBuilder(
-		t,
-		testutil.NewBalancer(testutil.WithInvokeHandlers(testutil.InvokeHandlers{
-			testutil.TableCreateSession: func(interface{}) (proto.Message, error) {
-				return &Ydb_Table.CreateSessionResult{
-					SessionId: testutil.SessionID(),
-				}, nil
-			},
-			testutil.TableDeleteSession: func(interface{}) (proto.Message, error) {
-				return &Ydb_Table.DeleteSessionResponse{}, nil
-			},
-		})),
-		3,
-		config.WithSizeLimit(3),
-		config.WithIdleThreshold(time.Hour),
-		config.WithTrace(
-			trace.Table{
-				OnPoolPut: func(info trace.TablePoolPutStartInfo) func(trace.TablePoolPutDoneInfo) {
-					wg.Add(1)
-					return func(info trace.TablePoolPutDoneInfo) {
-						wg.Done()
-					}
+	xtest.TestManyTimes(t, func(t testing.TB) {
+		p := newClientWithStubBuilder(
+			t,
+			testutil.NewBalancer(testutil.WithInvokeHandlers(testutil.InvokeHandlers{
+				testutil.TableCreateSession: func(interface{}) (proto.Message, error) {
+					return &Ydb_Table.CreateSessionResult{
+						SessionId: testutil.SessionID(),
+					}, nil
 				},
-				OnPoolSessionClose: func(
-					info trace.TablePoolSessionCloseStartInfo,
-				) func(
-					doneInfo trace.TablePoolSessionCloseDoneInfo,
-				) {
-					wg.Add(1)
-					return func(info trace.TablePoolSessionCloseDoneInfo) {
-						wg.Done()
-					}
+				testutil.TableDeleteSession: func(interface{}) (proto.Message, error) {
+					return &Ydb_Table.DeleteSessionResponse{}, nil
 				},
-			},
-		),
-	)
-	defer func() {
-		_ = p.Close(context.Background())
-	}()
-
-	var (
-		s1      = mustGetSession(t, p)
-		s2      = mustGetSession(t, p)
-		s3      = mustGetSession(t, p)
-		closed1 = false
-		closed2 = false
-		closed3 = false
-	)
-
-	s1.onClose = append(s1.onClose, func(s *session) { closed1 = true })
-	s2.onClose = append(s2.onClose, func(s *session) { closed2 = true })
-	s3.onClose = append(s3.onClose, func(s *session) { closed3 = true })
+			})),
+			3,
+			config.WithSizeLimit(3),
+			config.WithIdleThreshold(time.Hour),
+		)
+		defer func() {
+			_ = p.Close(context.Background())
+		}()
 
-	mustPutSession(t, p, s1)
-	mustPutSession(t, p, s2)
-	mustClose(t, p)
+		var (
+			s1      = mustGetSession(t, p)
+			s2      = mustGetSession(t, p)
+			s3      = mustGetSession(t, p)
+			closed1 = false
+			closed2 = false
+			closed3 = false
+		)
 
-	if !closed1 {
-		t.Fatalf("session1 was not closed")
-	}
-	if !closed2 {
-		t.Fatalf("session2 was not closed")
-	}
-	if closed3 {
-		t.Fatalf("unexpected session close")
-	}
+		s1.onClose = append(s1.onClose, func(s *session) { closed1 = true })
+		s2.onClose = append(s2.onClose, func(s *session) { closed2 = true })
+		s3.onClose = append(s3.onClose, func(s *session) { closed3 = true })
 
-	if err := p.Put(context.Background(), s3); !xerrors.Is(err, errClosedClient) {
-		t.Errorf(
-			"unexpected Put() error: %v; want %v",
-			err, errClosedClient,
-		)
-	}
-	wg.Wait()
+		mustPutSession(t, p, s1)
+		mustPutSession(t, p, s2)
+		mustClose(t, p)
 
-	if !closed3 {
-		t.Fatalf("session was not closed")
-	}
+		if !closed1 {
+			t.Errorf("session1 was not closed")
+		}
+		if !closed2 {
+			t.Errorf("session2 was not closed")
+		}
+		if !closed3 {
+			t.Errorf("session3 was not closed")
+		}
+	}, xtest.StopAfter(42*time.Second))
 }
 
 func TestRaceWgClosed(t *testing.T) {
@@ -380,7 +349,7 @@ func TestRaceWgClosed(t *testing.T) {
 		}
 		_ = p.Close(context.Background())
 		wg.Wait()
-	}, xtest.StopAfter(17*time.Second))
+	}, xtest.StopAfter(37*time.Second))
 }
 
 func TestSessionPoolDeleteReleaseWait(t *testing.T) {
@@ -1216,7 +1185,7 @@ func mustResetTimer(t *testing.T, ch <-chan time.Duration, exp time.Duration) {
 	}
 }
 
-func mustCreateSession(t *testing.T, p *Client) *session {
+func mustCreateSession(t testing.TB, p *Client) *session {
 	wg := sync.WaitGroup{}
 	defer wg.Wait()
 	s, err := p.internalPoolCreateSession(context.Background())
@@ -1227,7 +1196,7 @@ func mustCreateSession(t *testing.T, p *Client) *session {
 	return s
 }
 
-func mustGetSession(t *testing.T, p *Client) *session {
+func mustGetSession(t testing.TB, p *Client) *session {
 	wg := sync.WaitGroup{}
 	defer wg.Wait()
 	s, err := p.Get(context.Background())
@@ -1238,7 +1207,7 @@ func mustGetSession(t *testing.T, p *Client) *session {
 	return s
 }
 
-func mustPutSession(t *testing.T, p *Client, s *session) {
+func mustPutSession(t testing.TB, p *Client, s *session) {
 	wg := sync.WaitGroup{}
 	defer wg.Wait()
 	if err := p.Put(
@@ -1250,7 +1219,7 @@ func mustPutSession(t *testing.T, p *Client, s *session) {
 	}
 }
 
-func mustClose(t *testing.T, p *Client) {
+func mustClose(t testing.TB, p *Client) {
 	wg := sync.WaitGroup{}
 	defer wg.Wait()
 	if err := p.Close(context.Background()); err != nil {

internal/xerrors/stacktrace.go

@@ -28,12 +28,12 @@ func WithStackTrace(err error, opts ...withStackTraceOption) error {
 		o(&options)
 	}
 	return &stackError{
-		stackRecord: stackRecord(options.skipDepth + 1),
+		stackRecord: StackRecord(options.skipDepth + 1),
 		err:         err,
 	}
 }
 
-func stackRecord(depth int) string {
+func StackRecord(depth int) string {
 	function, file, line, _ := runtime.Caller(depth + 1)
 	name := runtime.FuncForPC(function).Name()
 	return name + "(" + fileName(file) + ":" + strconv.Itoa(line) + ")"