* Changed behavior on result.Err() on truncated result (return non-retryable error now)

Author: asmyasnikov

CHANGELOG.md

@@ -1,3 +1,4 @@
+* Changed behavior on `result.Err()` on truncated unary result (returns non-retryable error now)
 * Added simple transaction control constructors `table.OnlineReadOnlyTxControl()` and `table.StaleReadOnlyTxControl()`
 * Added transaction control specifier with context `ydb.WithTxControl`
 * Added value constructors `types.BytesValue`, `types.BytesValueFromString`, `types.TextValue`

internal/table/client.go

@@ -111,87 +111,96 @@ func (c *Client) createSession(ctx context.Context, opts ...createSessionOption)
 	if c.isClosed() {
 		return nil, errClosedClient
 	}
+
+	options := createSessionOptions{}
+	for _, o := range opts {
+		o(&options)
+	}
+
 	defer func() {
 		if s == nil {
 			return
 		}
-		options := createSessionOptions{}
-		for _, o := range opts {
-			o(&options)
-		}
 		for _, onCreate := range options.onCreate {
 			onCreate(s)
 		}
 		s.onClose = append(s.onClose, options.onClose...)
 	}()
-	type result struct {
-		s   *session
-		err error
-	}
-
-	ch := make(chan result)
 
-	c.spawnedGoroutines.Add(1)
-	go func() {
-		defer c.spawnedGoroutines.Done()
+	select {
+	case <-c.done:
+		return nil, xerrors.WithStackTrace(errClosedClient)
 
-		var (
+	default:
+		type result struct {
 			s   *session
 			err error
-		)
-
-		createSessionCtx := xcontext.WithoutDeadline(ctx)
-
-		if timeout := c.config.CreateSessionTimeout(); timeout > 0 {
-			var cancel context.CancelFunc
-			createSessionCtx, cancel = context.WithTimeout(createSessionCtx, timeout)
-			defer cancel()
 		}
 
-		s, err = c.build(createSessionCtx)
+		ch := make(chan result)
 
-		closeSession := func(s *session) {
-			if s == nil {
-				return
-			}
+		c.spawnedGoroutines.Add(1)
+		go func() {
+			defer c.spawnedGoroutines.Done()
+
+			var (
+				s   *session
+				err error
+			)
 
-			closeSessionCtx := xcontext.WithoutDeadline(ctx)
+			createSessionCtx := xcontext.WithoutDeadline(ctx)
 
-			if timeout := c.config.DeleteTimeout(); timeout > 0 {
+			if timeout := c.config.CreateSessionTimeout(); timeout > 0 {
 				var cancel context.CancelFunc
-				createSessionCtx, cancel = context.WithTimeout(closeSessionCtx, timeout)
+				createSessionCtx, cancel = context.WithTimeout(createSessionCtx, timeout)
 				defer cancel()
 			}
 
-			_ = s.Close(ctx)
-		}
+			s, err = c.build(createSessionCtx)
 
-		select {
-		case ch <- result{
-			s:   s,
-			err: err,
-		}: // nop
+			closeSession := func(s *session) {
+				if s == nil {
+					return
+				}
+
+				closeSessionCtx := xcontext.WithoutDeadline(ctx)
+
+				if timeout := c.config.DeleteTimeout(); timeout > 0 {
+					var cancel context.CancelFunc
+					createSessionCtx, cancel = context.WithTimeout(closeSessionCtx, timeout)
+					defer cancel()
+				}
+
+				_ = s.Close(ctx)
+			}
 
+			select {
+			case ch <- result{
+				s:   s,
+				err: err,
+			}: // nop
+
+			case <-c.done:
+				closeSession(s)
+
+			case <-ctx.Done():
+				closeSession(s)
+			}
+		}()
+
+		select {
 		case <-c.done:
-			closeSession(s)
+			return nil, xerrors.WithStackTrace(errClosedClient)
 
 		case <-ctx.Done():
-			closeSession(s)
-		}
-	}()
+			return nil, xerrors.WithStackTrace(ctx.Err())
 
-	select {
-	case r := <-ch:
-		if r.err != nil {
-			return nil, xerrors.WithStackTrace(r.err)
+		case r := <-ch:
+			if r.err != nil {
+				return nil, xerrors.WithStackTrace(r.err)
+			}
+			return r.s, nil
 		}
-		return r.s, nil
-
-	case <-c.done:
-		return nil, xerrors.WithStackTrace(errClosedClient)
-
-	case <-ctx.Done():
-		return nil, xerrors.WithStackTrace(ctx.Err())
 	}
 }
 

internal/table/scanner/result.go

@@ -98,6 +98,16 @@ func (r *baseResult) Reset(set *Ydb.ResultSet, columnNames ...string) {
 	}
 }
 
+func (r *unaryResult) Err() (err error) {
+	if err = r.baseResult.scanner.Err(); err != nil {
+		return err
+	}
+	if r.truncated() {
+		return xerrors.WithStackTrace(errTruncated)
+	}
+	return nil
+}
+
 func (r *unaryResult) NextResultSetErr(ctx context.Context, columns ...string) (err error) {
 	if r.isClosed() {
 		return xerrors.WithStackTrace(errAlreadyClosed)
@@ -114,6 +124,10 @@ func (r *unaryResult) NextResultSet(ctx context.Context, columns ...string) bool
 	return r.NextResultSetErr(ctx, columns...) == nil
 }
 
+func (r *streamResult) Err() (err error) {
+	return r.baseResult.scanner.Err()
+}
+
 func (r *streamResult) NextResultSetErr(ctx context.Context, columns ...string) (err error) {
 	if r.isClosed() {
 		return xerrors.WithStackTrace(errAlreadyClosed)

internal/table/scanner/scanner.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"database/sql"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"math"
@@ -22,6 +23,8 @@ import (
 	"github.com/ydb-platform/ydb-go-sdk/v3/table/types"
 )
 
+var errTruncated = xerrors.Wrap(errors.New("truncated result"))
+
 type scanner struct {
 	set       *Ydb.ResultSet
 	row       *Ydb.Value
@@ -206,11 +209,22 @@ func (s *scanner) Truncated() bool {
 	return s.set.Truncated
 }
 
+// Truncated returns true if current result set has been truncated by server
+func (s *scanner) truncated() bool {
+	if s.set == nil {
+		return false
+	}
+	return s.set.Truncated
+}
+
 // Err returns error caused Scanner to be broken.
 func (s *scanner) Err() error {
 	s.errMtx.RLock()
 	defer s.errMtx.RUnlock()
-	return s.err
+	if s.err != nil {
+		return s.err
+	}
+	return nil
 }
 
 // Must not be exported.

table/table_e2e_test.go

@@ -1510,7 +1510,7 @@ func TestLongStream(t *testing.T) {
 					time.Sleep(discoveryInterval)
 				}
 				if err = res.Err(); err != nil {
-					return fmt.Errorf("received error: %w (duration: %v)", err, time.Since(start))
+					return fmt.Errorf("received error (duration: %v): %w", time.Since(start), err)
 				}
 				if rowsCount != upsertRowsCount {
 					return fmt.Errorf("wrong rows count: %v, expected: %d (duration: %v)",
@@ -1551,7 +1551,7 @@ func TestLongStream(t *testing.T) {
 					time.Sleep(discoveryInterval)
 				}
 				if err = res.Err(); err != nil {
-					return fmt.Errorf("received error: %w (duration: %v)", err, time.Since(start))
+					return fmt.Errorf("received error (duration: %v): %w", time.Since(start), err)
 				}
 				if rowsCount != upsertRowsCount {
 					return fmt.Errorf("wrong rows count: %v, expected: %d (duration: %v)",