Merge branch 'master' into back-comp

Author: size12

CHANGELOG.md

@@ -1,5 +1,8 @@
-* Added `Struct` support for `Variant` in `ydb.ParamsBuilder()`
 * Restored `WithSessionPoolKeepAliveMinSize` and `WithSessionPoolKeepAliveTimeout` for backward compatibility.
+* Fixed leak timers
+* Changed default StartTime (time of retries for connect to server) for topic writer from 1 minute to infinite (can be overrided by WithWriterStartTimeout topic option)
+* Added `Struct` support for `Variant` in `ydb.ParamsBuilder()`
+* Added `go` with anonymous function case in `gstack`
 
 ## v3.61.2
 * Changed default transaction control to `NoTx` for execute query through query service client 

internal/cmd/gstack/main.go

@@ -79,6 +79,12 @@ func getCallExpressionsFromStmt(statement ast.Stmt) (listOfCallExpressions []*as
 		body = stmt.Body
 	case *ast.ForStmt:
 		body = stmt.Body
+	case *ast.GoStmt:
+		if fun, ok := stmt.Call.Fun.(*ast.FuncLit); ok {
+			listOfCallExpressions = append(listOfCallExpressions, getListOfCallExpressionsFromBlockStmt(fun.Body)...)
+		} else {
+			listOfCallExpressions = append(listOfCallExpressions, stmt.Call)
+		}
 	case *ast.RangeStmt:
 		body = stmt.Body
 	case *ast.DeclStmt:

internal/coordination/session.go

@@ -139,10 +139,12 @@ func (s *session) newStream(
 
 		var client Ydb_Coordination_V1.CoordinationService_SessionClient
 		if lastChance {
+			timer := time.NewTimer(s.options.SessionKeepAliveTimeout)
 			select {
-			case <-time.After(s.options.SessionKeepAliveTimeout):
+			case <-timer.C:
 			case client = <-result:
 			}
+			timer.Stop()
 
 			if client != nil {
 				return client, nil
@@ -175,10 +177,12 @@ func (s *session) newStream(
 		}
 
 		// Waiting for some time before trying to reconnect.
+		sessionReconnectDelay := time.NewTimer(s.options.SessionReconnectDelay)
 		select {
-		case <-time.After(s.options.SessionReconnectDelay):
+		case <-sessionReconnectDelay.C:
 		case <-s.ctx.Done():
 		}
+		sessionReconnectDelay.Stop()
 
 		if s.ctx.Err() != nil {
 			// Give this session the last chance to stop gracefully if the session is canceled in the reconnect cycle.
@@ -247,6 +251,7 @@ func (s *session) mainLoop(path string, sessionStartedChan chan struct{}) {
 
 		// Wait for the session started response unless the stream context is done. We intentionally do not take into
 		// account stream context cancellation in order to proceed with the graceful shutdown if it requires reconnect.
+		sessionStartTimer := time.NewTimer(s.options.SessionStartTimeout)
 		select {
 		case start := <-sessionStarted:
 			trace.CoordinationOnSessionStarted(s.client.config.Trace(), start.GetSessionId(), s.sessionID)
@@ -258,13 +263,14 @@ func (s *session) mainLoop(path string, sessionStartedChan chan struct{}) {
 				cancelStream()
 			}
 			close(startSending)
-		case <-time.After(s.options.SessionStartTimeout):
+		case <-sessionStartTimer.C:
 			// Reconnect if no response was received before the timeout occurred.
 			trace.CoordinationOnSessionStartTimeout(s.client.config.Trace(), s.options.SessionStartTimeout)
 			cancelStream()
 		case <-streamCtx.Done():
 		case <-s.ctx.Done():
 		}
+		sessionStartTimer.Stop()
 
 		for {
 			// Respect the failure reason priority: if the session context is done, we must stop the session, even
@@ -280,8 +286,9 @@ func (s *session) mainLoop(path string, sessionStartedChan chan struct{}) {
 			}
 
 			keepAliveTime := time.Until(s.getLastGoodResponseTime().Add(s.options.SessionKeepAliveTimeout))
+			keepAliveTimeTimer := time.NewTimer(keepAliveTime)
 			select {
-			case <-time.After(keepAliveTime):
+			case <-keepAliveTimeTimer.C:
 				last := s.getLastGoodResponseTime()
 				if time.Since(last) > s.options.SessionKeepAliveTimeout {
 					// Reconnect if the underlying stream is likely to be dead.
@@ -295,6 +302,7 @@ func (s *session) mainLoop(path string, sessionStartedChan chan struct{}) {
 			case <-streamCtx.Done():
 			case <-s.ctx.Done():
 			}
+			keepAliveTimeTimer.Stop()
 		}
 
 		if closing {
@@ -318,8 +326,10 @@ func (s *session) mainLoop(path string, sessionStartedChan chan struct{}) {
 			)
 
 			// Wait for the session stopped response unless the stream context is done.
+			sessionStopTimeout := time.NewTimer(s.options.SessionStopTimeout)
 			select {
 			case stop := <-sessionStopped:
+				sessionStopTimeout.Stop()
 				trace.CoordinationOnSessionStopped(s.client.config.Trace(), stop.GetSessionId(), s.sessionID)
 				if stop.GetSessionId() == s.sessionID {
 					cancelStream()
@@ -329,15 +339,19 @@ func (s *session) mainLoop(path string, sessionStartedChan chan struct{}) {
 
 				// Reconnect if the server response is invalid.
 				cancelStream()
-			case <-time.After(s.options.SessionStopTimeout):
+			case <-sessionStopTimeout.C:
+				sessionStopTimeout.Stop() // no really need, call stop for common style only
+
 				// Reconnect if no response was received before the timeout occurred.
 				trace.CoordinationOnSessionStopTimeout(s.client.config.Trace(), s.options.SessionStopTimeout)
 				cancelStream()
 			case <-s.ctx.Done():
+				sessionStopTimeout.Stop()
 				cancelStream()
 
 				return
 			case <-streamCtx.Done():
+				sessionStopTimeout.Stop()
 			}
 		}
 

internal/stack/function_id_test.go

@@ -1,7 +1,9 @@
 package stack
 
 import (
+	"sync"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -22,6 +24,22 @@ func (e *starType) starredCall() string {
 	return FunctionID("").FunctionID()
 }
 
+func anonymousFunctionCall() string {
+	var result string
+	var mu sync.Mutex
+	go func() {
+		mu.Lock()
+		defer mu.Unlock()
+		result = FunctionID("").FunctionID()
+	}()
+	time.Sleep(time.Second)
+
+	mu.Lock()
+	defer mu.Unlock()
+
+	return result
+}
+
 func TestFunctionIDForGenericType(t *testing.T) {
 	t.Run("StaticFunc", func(t *testing.T) {
 		require.Equal(t,
@@ -42,4 +60,10 @@ func TestFunctionIDForGenericType(t *testing.T) {
 			x.starredCall(),
 		)
 	})
+	t.Run("AnonymousFunctionCall", func(t *testing.T) {
+		require.Equal(t,
+			"github.com/ydb-platform/ydb-go-sdk/v3/internal/stack.anonymousFunctionCall",
+			anonymousFunctionCall(),
+		)
+	})
 }

internal/table/client.go

@@ -489,7 +489,10 @@ func (c *Client) internalPoolWaitFromCh(ctx context.Context, t *trace.Table) (s
 
 	var createSessionTimeoutCh <-chan time.Time
 	if timeout := c.config.CreateSessionTimeout(); timeout > 0 {
-		createSessionTimeoutCh = c.clock.After(timeout)
+		createSessionTimeoutChTimer := c.clock.NewTimer(timeout)
+		defer createSessionTimeoutChTimer.Stop()
+
+		createSessionTimeoutCh = createSessionTimeoutChTimer.Chan()
 	}
 
 	select {

internal/topic/retriable_error.go

@@ -13,7 +13,7 @@ import (
 )
 
 const (
-	DefaultStartTimeout          = time.Minute
+	DefaultStartTimeout          = value.InfiniteDuration
 	connectionEstablishedTimeout = time.Minute
 )
 

internal/topic/topicreaderinternal/committer.go

@@ -166,7 +166,10 @@ func (c *committer) waitSendTrigger(ctx context.Context) {
 		return
 	}
 
-	finish := c.clock.After(c.BufferTimeLagTrigger)
+	bufferTimeLagTriggerTimer := c.clock.NewTimer(c.BufferTimeLagTrigger)
+	defer bufferTimeLagTriggerTimer.Stop()
+
+	finish := bufferTimeLagTriggerTimer.Chan()
 	if c.BufferCountTrigger == 0 {
 		select {
 		case <-ctxDone:

internal/topic/topicreaderinternal/stream_reconnector.go

@@ -328,9 +328,12 @@ func (r *readerReconnector) connectWithTimeout() (_ batchedStreamReader, err err
 		result <- connectResult{stream: stream, err: err}
 	}()
 
+	connectionTimoutTimer := r.clock.NewTimer(r.connectTimeout)
+	defer connectionTimoutTimer.Stop()
+
 	var res connectResult
 	select {
-	case <-r.clock.After(r.connectTimeout):
+	case <-connectionTimoutTimer.Chan():
 		// cancel connection context only if timeout exceed while connection
 		// because if cancel context after connect - it will break
 		cancel()

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -380,16 +380,21 @@ func (w *WriterReconnector) connectionLoop(ctx context.Context) {
 		prevAttemptTime = now
 
 		if reconnectReason != nil {
-			if backoff, retry := topic.CheckRetryMode(reconnectReason, w.retrySettings, w.clock.Since(startOfRetries)); retry {
+			retryDuration := w.clock.Since(startOfRetries)
+			if backoff, retry := topic.CheckRetryMode(reconnectReason, w.retrySettings, retryDuration); retry {
 				delay := backoff.Delay(attempt)
+				delayTimer := w.clock.NewTimer(delay)
 				select {
 				case <-doneCtx:
+					delayTimer.Stop()
+
 					return
-				case <-w.clock.After(delay):
+				case <-delayTimer.Chan():
+					delayTimer.Stop() // no really need, stop for common style only
 					// pass
 				}
 			} else {
-				_ = w.close(ctx, reconnectReason)
+				_ = w.close(ctx, fmt.Errorf("%w, was retried (%v)", reconnectReason, retryDuration))
 
 				return
 			}

internal/xsql/connector.go

@@ -262,10 +262,14 @@ func (c *Connector) idleCloser() (idleStopper func()) {
 	ctx, idleStopper = xcontext.WithCancel(context.Background())
 	go func() {
 		for {
+			idleThresholdTimer := c.clock.NewTimer(c.idleThreshold)
 			select {
 			case <-ctx.Done():
+				idleThresholdTimer.Stop()
+
 				return
-			case <-c.clock.After(c.idleThreshold):
+			case <-idleThresholdTimer.Chan():
+				idleThresholdTimer.Stop() // no really need, stop for common style only
 				c.connsMtx.RLock()
 				conns := make([]*conn, 0, len(c.conns))
 				for cc := range c.conns {