resolve issue #66

* retry tarantool request only on some vshard errors
* do not export anymore storageCallAssertError, bucketStatError

Author: nurzhan-saktaganov

CHANGELOG.md

@@ -17,6 +17,7 @@ FEATURES:
 * All discovering logs has new prefix [DISCOVERY]
 * Introduce Replicaset.CallAsync, it is usefull to send concurrent requests to replicasets;
 	additionally, CallAsync provides new interface to interact with replicaset without cons of interface of ReplicaCall
+* Retry tarantool request only on some vshard errors (#66)
 
 REFACTOR:
 
@@ -28,6 +29,7 @@ REFACTOR:
 * BucketStat: split into bucketStatAsync and bucketStatWait parts
 * BucketDiscovery: do not spawn goroutines, just use futures in the single goroutine
 * BucketResolve: make it alias for BucketDiscovery
+* do not export anymore storageCallAssertError, bucketStatError
 
 TESTS:
 

api.go

@@ -27,16 +27,17 @@ func (c VshardMode) String() string {
 	return string(c)
 }
 
-type StorageCallAssertError struct {
+type storageCallAssertError struct {
 	Code     int         `msgpack:"code"`
 	BaseType string      `msgpack:"base_type"`
 	Type     string      `msgpack:"type"`
 	Message  string      `msgpack:"message"`
 	Trace    interface{} `msgpack:"trace"`
 }
 
-func (s StorageCallAssertError) Error() string {
-	return fmt.Sprintf("vshard.storage.call assert error code: %d, type:%s, message: %s", s.Code, s.Type, s.Message)
+func (s storageCallAssertError) Error() string {
+	type alias storageCallAssertError
+	return fmt.Sprintf("%+v", alias(s))
 }
 
 type StorageCallVShardError struct {
@@ -51,7 +52,11 @@ type StorageCallVShardError struct {
 }
 
 func (s StorageCallVShardError) Error() string {
-	return fmt.Sprintf("vshard.storage.call bucket error bucket_id: %d, reason: %s, name: %s", s.BucketID, s.Reason, s.Name)
+	// Just print struct as is, use hack with alias type to avoid recursion:
+	// %v attempts to call Error() method for s, which is recursion.
+	// This alias doesn't have method Error().
+	type alias StorageCallVShardError
+	return fmt.Sprintf("%+v", alias(s))
 }
 
 type StorageResultTypedFunc = func(result interface{}) error
@@ -76,6 +81,8 @@ func (r *Router) RouterCallImpl(ctx context.Context,
 	fnc string,
 	args interface{}) (interface{}, StorageResultTypedFunc, error) {
 
+	const vshardStorageClientCall = "vshard.storage.call"
+
 	if bucketID < 1 || r.cfg.TotalBucketCount < bucketID {
 		return nil, nil, fmt.Errorf("bucket id is out of range: %d (total %d)", bucketID, r.cfg.TotalBucketCount)
 	}
@@ -87,7 +94,7 @@ func (r *Router) RouterCallImpl(ctx context.Context,
 	timeout := opts.Timeout
 	timeStart := time.Now()
 
-	req := tarantool.NewCallRequest("vshard.storage.call")
+	req := tarantool.NewCallRequest(vshardStorageClientCall)
 	req = req.Context(ctx)
 	req = req.Args([]interface{}{
 		bucketID,
@@ -97,6 +104,7 @@ func (r *Router) RouterCallImpl(ctx context.Context,
 	})
 
 	var err error
+	var vshardError StorageCallVShardError
 
 	for {
 		if since := time.Since(timeStart); since > timeout {
@@ -114,10 +122,9 @@ func (r *Router) RouterCallImpl(ctx context.Context,
 
 		rs, err = r.BucketResolve(ctx, bucketID)
 		if err != nil {
-			r.log().Debugf(ctx, "cant resolve bucket %d with error: %s", bucketID, err.Error())
-
 			r.metrics().RetryOnCall("bucket_resolve_error")
-			continue
+
+			return nil, nil, fmt.Errorf("cant resolve bucket %d: %w", bucketID, err)
 		}
 
 		r.log().Infof(ctx, "try call %s on replicaset %s for bucket %d", fnc, rs.info.Name, bucketID)
@@ -127,97 +134,73 @@ func (r *Router) RouterCallImpl(ctx context.Context,
 		var respData []interface{}
 		respData, err = future.Get()
 		if err != nil {
-			r.log().Errorf(ctx, "got future error: %s", err)
 			r.metrics().RetryOnCall("future_get_error")
 
-			continue
+			return nil, nil, fmt.Errorf("got error on future.Get(): %w", err)
 		}
 
-		r.log().Debugf(ctx, "got call result response data %s", respData)
+		r.log().Debugf(ctx, "got call result response data %v", respData)
 
-		if len(respData) < 1 {
+		if len(respData) == 0 {
 			// vshard.storage.call(func) returns up to two values:
-			// - true/false
+			// - true/false/nil
 			// - func result, omitted if func does not return anything
-			err = fmt.Errorf("invalid length of response data: must be >= 1, current: %d", len(respData))
-
-			r.log().Errorf(ctx, "%s", err.Error())
-
 			r.metrics().RetryOnCall("resp_data_error")
-			continue
+
+			return nil, nil, fmt.Errorf("protocol violation %s: got empty response", vshardStorageClientCall)
 		}
 
 		if respData[0] == nil {
-			vshardErr := &StorageCallVShardError{}
-
-			if len(respData) < 2 {
-				err = fmt.Errorf("unexpected response length when respData[0] == nil: %d", len(respData))
-			} else {
-				err = mapstructure.Decode(respData[1], vshardErr)
+			if len(respData) != 2 {
+				return nil, nil, fmt.Errorf("protocol violation %s: length is %d when respData[0] is nil", vshardStorageClientCall, len(respData))
 			}
 
+			err = mapstructure.Decode(respData[1], &vshardError)
 			if err != nil {
+				// Something unexpected happened: we couldn't decode respData[1] as a vshardError,
+				// so return reason why and respData[1], that is supposed to be a vshardError.
 				r.metrics().RetryOnCall("internal_error")
 
-				err = fmt.Errorf("cant decode vhsard err by trarantool with err: %s; continue try", err)
-
-				r.log().Errorf(ctx, "%s", err.Error())
-				continue
+				return nil, nil, fmt.Errorf("cant decode vhsard err by trarantool with err: %v (%v)", err, respData[1])
 			}
 
-			err = vshardErr
-
-			r.log().Errorf(ctx, "got vshard storage call error: %s", err)
-
-			if vshardErr.Name == "WRONG_BUCKET" ||
-				vshardErr.Name == "BUCKET_IS_LOCKED" ||
-				vshardErr.Name == "TRANSFER_IS_IN_PROGRESS" {
+			switch vshardError.Name {
+			case "WRONG_BUCKET", "BUCKET_IS_LOCKED", "TRANSFER_IS_IN_PROGRESS":
 				r.BucketReset(bucketID)
 				r.metrics().RetryOnCall("bucket_migrate")
 
-				r.log().Debugf(ctx, "retrying cause bucket in migrate state (%s)", vshardErr.Name)
+				r.log().Debugf(ctx, "retrying %s cause bucket in migrate state (%s)", fnc, vshardError.Error())
 
+				// this vshardError will be returned to a caller in case of timeout
+				err = &vshardError
 				continue
 			}
 
-			continue
+			return nil, nil, &vshardError
 		}
 
-		isVShardRespOk := false
+		var isVShardRespOk bool
 		err = future.GetTyped(&[]interface{}{&isVShardRespOk})
 		if err != nil {
-			r.log().Debugf(ctx, "cant get typed with err: %s", err)
-
-			continue
+			return nil, nil, fmt.Errorf("protocol violation %s: can't decode respData[0] as boolean: %v", vshardStorageClientCall, err)
 		}
 
-		if !isVShardRespOk { // error
-			errorResp := &StorageCallAssertError{}
-
-			// Since we got respData[0] == false, it means that assert has happened
+		if !isVShardRespOk {
+			// Since we got respData[0] == false, it means that an error has happened
 			// while executing user-defined function on vshard storage.
 			// In this case, vshard storage must return a pair: false, error.
-			if len(respData) < 2 {
-				err = fmt.Errorf("protocol violation: unexpected response length when respData[0] == false: %d", len(respData))
-			} else {
-				err = future.GetTyped(&[]interface{}{&isVShardRespOk, errorResp})
+			if len(respData) != 2 {
+				return nil, nil, fmt.Errorf("protocol violation %s: response length is %d when respData[0] is false", vshardStorageClientCall, len(respData))
 			}
 
+			var assertError storageCallAssertError
+			err = mapstructure.Decode(respData[1], &assertError)
 			if err != nil {
-				// Either protocol has been violated or decoding has failed.
-				err = fmt.Errorf("cant get typed vshard err with err: %s", err)
-			} else {
-				// StorageCallAssertError successfully has been decoded.
-				err = errorResp
+				// We could not decode respData[1] as assertError, so return respData[1] as is, add info why we could not decode.
+				return nil, nil, fmt.Errorf("%s: %s failed %v (decoding to assertError failed %v)", vshardStorageClientCall, fnc, respData[1], err)
 			}
 
-			if errorResp.Type == "ClientError" || errorResp.Type == "LuajitError" {
-				return nil, nil, errorResp
-			}
-
-			r.log().Debugf(ctx, "retry cause vhsard response not ok: %s", errorResp)
-
-			continue
+			return nil, nil, fmt.Errorf("%s: %s failed: %+v", vshardStorageClientCall, fnc, assertError)
 		}
 
 		r.metrics().RequestDuration(time.Since(timeStart), true, false)
@@ -363,14 +346,14 @@ func (r *Router) RouterMapCallRWImpl(
 				return nil, fmt.Errorf("protocol violation: invalid respData length when respData[0] == nil, must be = 2, current: %d", len(respData))
 			}
 
-			assertError := &StorageCallAssertError{}
-			err = mapstructure.Decode(respData[1], assertError)
+			var assertError storageCallAssertError
+			err = mapstructure.Decode(respData[1], &assertError)
 			if err != nil {
-				// TODO: not only StorageCallAssertError is possible here?
+				// We could not decode respData[1] as assertError, so return respData[1] as is, add info why we could not decode.
 				return nil, fmt.Errorf("storage_map failed on %v: %+v (decoding to assertError failed %v)", rsFuture.uuid, respData[1], err)
 			}
 
-			return nil, fmt.Errorf("storage_map failed on %v: %w", rsFuture.uuid, assertError)
+			return nil, fmt.Errorf("storage_map failed on %v: %+v", rsFuture.uuid, assertError)
 		}
 
 		var isVShardRespOk bool

api_test.go

@@ -68,6 +68,6 @@ func TestRouter_RouterCallImpl(t *testing.T) {
 		})
 
 		_, _, err := r.RouterCallImpl(ctx, 5, CallOpts{Timeout: time.Second}, "test", []byte("test"))
-		require.ErrorIs(t, futureError, err)
+		require.ErrorIs(t, err, futureError)
 	})
 }

error.go

@@ -224,7 +224,7 @@ var Errors = map[int]Error{
 	},
 }
 
-type BucketStatError struct {
+type bucketStatError struct {
 	BucketID uint64 `msgpack:"bucket_id"`
 	Reason   string `msgpack:"reason"`
 	Code     int    `msgpack:"code"`
@@ -233,6 +233,7 @@ type BucketStatError struct {
 	Name     string `msgpack:"name"`
 }
 
-func (bse BucketStatError) Error() string {
-	return "todo"
+func (bse bucketStatError) Error() string {
+	type alias bucketStatError
+	return fmt.Sprintf("%+v", alias(bse))
 }

replicaset.go

@@ -60,22 +60,20 @@ func bucketStatWait(future *tarantool.Future) (BucketStatInfo, error) {
 		return bsInfo, err
 	}
 
-	if len(respData) < 1 {
-		return bsInfo, fmt.Errorf("respData len is 0 for bucketStatWait; unsupported or broken proto")
+	if len(respData) == 0 {
+		return bsInfo, fmt.Errorf("protocol violation bucketStatWait: empty response")
 	}
 
 	if respData[0] == nil {
-
-		if len(respData) < 2 {
-			return bsInfo, fmt.Errorf("respData len < 2 when respData[0] is nil for bucketStatWait")
+		if len(respData) != 2 {
+			return bsInfo, fmt.Errorf("protocol violation bucketStatWait: invalid response length %d when respData[0] is nil", len(respData))
 		}
 
-		var tmp interface{} // todo: fix non-panic crutch
-		bsError := &BucketStatError{}
-
-		err := future.GetTyped(&[]interface{}{tmp, bsError})
+		var bsError bucketStatError
+		err = mapstructure.Decode(respData[1], &bsError)
 		if err != nil {
-			return bsInfo, err
+			// We could not decode respData[1] as bsError, so return respData[1] as is, add info why we could not decode.
+			return bsInfo, fmt.Errorf("bucketStatWait error: %v (can't decode into bsError: %v)", respData[1], err)
 		}
 
 		return bsInfo, bsError
@@ -106,42 +104,28 @@ func (rs *Replicaset) ReplicaCall(
 		timeout = opts.Timeout
 	}
 
-	timeStart := time.Now()
-
 	ctx, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 
 	req := tarantool.NewCallRequest(fnc).
 		Context(ctx).
 		Args(args)
 
-	var (
-		respData []interface{}
-		err      error
-	)
-
-	for {
-		if since := time.Since(timeStart); since > timeout {
-			return nil, nil, err
-		}
-
-		future := rs.conn.Do(req, opts.PoolMode)
-
-		respData, err = future.Get()
-		if err != nil {
-			continue
-		}
+	future := rs.conn.Do(req, opts.PoolMode)
 
-		if len(respData) == 0 {
-			// Since this method returns the first element of respData by contract, we can't return anything is this case (broken interface)
-			err = fmt.Errorf("response data is empty")
-			continue
-		}
+	respData, err := future.Get()
+	if err != nil {
+		return nil, nil, fmt.Errorf("got error on future.Get(): %w", err)
+	}
 
-		return respData[0], func(result interface{}) error {
-			return future.GetTyped(&[]interface{}{&result})
-		}, nil
+	if len(respData) == 0 {
+		// Since this method returns the first element of respData by contract, we can't return anything is this case (broken interface)
+		return nil, nil, fmt.Errorf("%s response data is empty", fnc)
 	}
+
+	return respData[0], func(result interface{}) error {
+		return future.GetTyped(&[]interface{}{&result})
+	}, nil
 }
 
 // Call sends async request to remote storage

tests/tnt/replicaset_test.go

@@ -2,7 +2,6 @@ package tnt
 
 import (
 	"context"
-	"log"
 	"testing"
 	"time"
 
@@ -14,10 +13,7 @@ import (
 )
 
 func TestReplicasetReplicaCall(t *testing.T) {
-	if !isCorrectRun() {
-		log.Printf("Incorrect run of tnt-test framework")
-		return
-	}
+	skipOnInvalidRun(t)
 
 	t.Parallel()
 
@@ -101,10 +97,7 @@ func TestReplicasetReplicaCall(t *testing.T) {
 }
 
 func TestReplicsetCallAsync(t *testing.T) {
-	if !isCorrectRun() {
-		log.Printf("Incorrect run of tnt-test framework")
-		return
-	}
+	skipOnInvalidRun(t)
 
 	t.Parallel()
 

tests/tnt/router_call_test.go

@@ -57,10 +57,10 @@ func TestRouterCallProto(t *testing.T) {
 	require.NotNil(t, err, "RouterCallImpl echo finised with nil args")
 
 	_, _, err = router.RouterCallImpl(ctx, bucketID, callOpts, "raise_luajit_error", args)
-	require.IsType(t, &vshardrouter.StorageCallAssertError{}, err, "RouterCallImpl raise_luajit_error finished with StorageCallAssertError")
+	require.NotNil(t, err, "RouterCallImpl raise_luajit_error finished with err")
 
 	_, _, err = router.RouterCallImpl(ctx, bucketID, callOpts, "raise_client_error", args)
-	require.IsType(t, &vshardrouter.StorageCallAssertError{}, err, "RouterCallImpl raise_client_error finished with StorageCallAssertError")
+	require.NotNil(t, err, "RouterCallImpl raise_client_error finished with err")
 
 	// maxRespLen is due to:
 	// https://github.com/tarantool/vshard/blob/dfa2cc8a2aff221d5f421298851a9a229b2e0434/vshard/storage/init.lua#L3130