* Added option ydb.WithSessionPoolSessionIdleTimeToLive for restric…

…t idle time of query sessions

* Fixed bug with leak of query transactions

CHANGELOG.md

@@ -1,3 +1,5 @@
+* Added option `ydb.WithSessionPoolSessionIdleTimeToLive` for restrict idle time of query sessions
+* Fixed bug with leak of query transactions
 * Changed `ydb_go_sdk_ydb_driver_conn_requests` metrics splitted to `ydb_go_sdk_ydb_driver_conn_request_statuses` and `ydb_go_sdk_ydb_driver_conn_request_methods`
 * Fixed metadata for operation service connection
 * Fixed composing query traces in call `db.Query.Do[Tx]` using option `query.WithTrace`

internal/pool/pool.go

@@ -26,18 +26,18 @@ type (
 		Item
 	}
 	Config[PT ItemConstraint[T], T any] struct {
-		trace         *Trace
-		clock         clockwork.Clock
-		limit         int
-		createTimeout time.Duration
-		createItem    func(ctx context.Context) (PT, error)
-		closeTimeout  time.Duration
-		closeItem     func(ctx context.Context, item PT)
-		idleThreshold time.Duration
+		trace          *Trace
+		clock          clockwork.Clock
+		limit          int
+		createTimeout  time.Duration
+		createItem     func(ctx context.Context) (PT, error)
+		closeTimeout   time.Duration
+		closeItem      func(ctx context.Context, item PT)
+		idleTimeToLive time.Duration
 	}
 	itemInfo[PT ItemConstraint[T], T any] struct {
-		idle    *xlist.Element[PT]
-		touched time.Time
+		idle      *xlist.Element[PT]
+		lastUsage time.Time
 	}
 	waitChPool[PT ItemConstraint[T], T any] interface {
 		GetOrNew() *chan PT
@@ -99,9 +99,9 @@ func WithTrace[PT ItemConstraint[T], T any](t *Trace) Option[PT, T] {
 	}
 }
 
-func WithIdleThreshold[PT ItemConstraint[T], T any](idleThreshold time.Duration) Option[PT, T] {
+func WithIdleTimeToLive[PT ItemConstraint[T], T any](idleTTL time.Duration) Option[PT, T] {
 	return func(c *Config[PT, T]) {
-		c.idleThreshold = idleThreshold
+		c.idleTimeToLive = idleTTL
 	}
 }
 
@@ -218,7 +218,7 @@ func makeAsyncCreateItemFunc[PT ItemConstraint[T], T any]( //nolint:funlen
 			if newItem != nil {
 				p.mu.WithLock(func() {
 					p.index[newItem] = itemInfo[PT, T]{
-						touched: p.config.clock.Now(),
+						lastUsage: p.config.clock.Now(),
 					}
 				})
 			}
@@ -461,7 +461,7 @@ func (p *Pool[PT, T]) peekFirstIdle() (item PT, touched time.Time) {
 		panic(fmt.Sprintf("inconsistent index: (%v, %+v, %+v)", has, el, info.idle))
 	}
 
-	return item, info.touched
+	return item, info.lastUsage
 }
 
 // removes first session from idle and resets the keepAliveCount
@@ -547,7 +547,7 @@ func (p *Pool[PT, T]) pushIdle(item PT, now time.Time) {
 	}
 
 	p.changeState(func() Stats {
-		info.touched = now
+		info.lastUsage = now
 		info.idle = p.idle.PushBack(item)
 		p.index[item] = info
 
@@ -595,7 +595,7 @@ func (p *Pool[PT, T]) getItem(ctx context.Context) (item PT, finalErr error) { /
 					return info
 				})
 
-				if p.config.idleThreshold > 0 && p.config.clock.Since(info.touched) > p.config.idleThreshold {
+				if p.config.idleTimeToLive > 0 && p.config.clock.Since(info.lastUsage) > p.config.idleTimeToLive {
 					p.closeItem(ctx, item)
 					p.mu.WithLock(func() {
 						p.changeState(func() Stats {

internal/pool/pool_test.go

@@ -394,22 +394,22 @@ func TestPool(t *testing.T) {
 					// replace default async closer for sync testing
 					WithSyncCloseItem[*testItem, testItem](),
 					WithClock[*testItem, testItem](fakeClock),
-					WithIdleThreshold[*testItem, testItem](idleThreshold),
+					WithIdleTimeToLive[*testItem, testItem](idleThreshold),
 					WithTrace[*testItem, testItem](defaultTrace),
 				)
 
 				s1 := mustGetItem(t, p)
 				s2 := mustGetItem(t, p)
 
 				// Put both items at the absolutely same time.
-				// That is, both items must be updated their touched timestamp.
+				// That is, both items must be updated their lastUsage timestamp.
 				mustPutItem(t, p, s1)
 				mustPutItem(t, p, s2)
 
 				require.Len(t, p.index, 2)
 				require.Equal(t, 2, p.idle.Len())
 
-				// Move clock to longer than idleThreshold
+				// Move clock to longer than idleTimeToLive
 				fakeClock.Advance(idleThreshold + time.Nanosecond)
 
 				// on get item from idle list the pool must check the item idle timestamp
@@ -423,15 +423,15 @@ func TestPool(t *testing.T) {
 					t.Fatal("unexpected number of closed items")
 				}
 
-				// Move time to idleThreshold / 2 - this emulate a "spent" some time working within item.
+				// Move time to idleTimeToLive / 2 - this emulate a "spent" some time working within item.
 				fakeClock.Advance(idleThreshold / 2)
 
 				// Now put that item back
-				// pool must update a touched timestamp of item
+				// pool must update a lastUsage timestamp of item
 				mustPutItem(t, p, s3)
 
-				// Move time to idleThreshold / 2
-				// Total time since last updating touched timestampe is more than idleThreshold
+				// Move time to idleTimeToLive / 2
+				// Total time since last updating lastUsage timestampe is more than idleTimeToLive
 				fakeClock.Advance(idleThreshold/2 + time.Nanosecond)
 
 				require.Len(t, p.index, 1)

internal/query/client.go

@@ -205,9 +205,7 @@ func do(
 
 		err := op(ctx, s)
 		if err != nil {
-			if xerrors.IsOperationError(err) {
-				s.SetStatus(session.StatusClosed)
-			}
+			s.SetStatus(session.StatusError)
 
 			return xerrors.WithStackTrace(err)
 		}
@@ -263,27 +261,27 @@ func doTx(
 	txSettings tx.Settings,
 	opts ...retry.Option,
 ) (finalErr error) {
-	err := do(ctx, pool, func(ctx context.Context, s *Session) (err error) {
+	err := do(ctx, pool, func(ctx context.Context, s *Session) (opErr error) {
 		tx, err := s.Begin(ctx, txSettings)
 		if err != nil {
 			return xerrors.WithStackTrace(err)
 		}
-		err = op(ctx, tx)
-		if err != nil {
-			errRollback := tx.Rollback(ctx)
-			if errRollback != nil {
-				return xerrors.WithStackTrace(xerrors.Join(err, errRollback))
+
+		defer func() {
+			_ = tx.Rollback(ctx)
+
+			if opErr != nil {
+				s.SetStatus(session.StatusError)
 			}
+		}()
 
+		err = op(ctx, tx)
+		if err != nil {
 			return xerrors.WithStackTrace(err)
 		}
+
 		err = tx.CommitTx(ctx)
 		if err != nil {
-			errRollback := tx.Rollback(ctx)
-			if errRollback != nil {
-				return xerrors.WithStackTrace(xerrors.Join(err, errRollback))
-			}
-
 			return xerrors.WithStackTrace(err)
 		}
 
@@ -530,6 +528,7 @@ func New(ctx context.Context, cc grpc.ClientConnInterface, cfg *config.Config) *
 			pool.WithTrace[*Session, Session](poolTrace(cfg.Trace())),
 			pool.WithCreateItemTimeout[*Session, Session](cfg.SessionCreateTimeout()),
 			pool.WithCloseItemTimeout[*Session, Session](cfg.SessionDeleteTimeout()),
+			pool.WithIdleTimeToLive[*Session, Session](cfg.SessionIdleTimeToLive()),
 			pool.WithCreateItemFunc(func(ctx context.Context) (_ *Session, err error) {
 				var (
 					createCtx    context.Context