Merge pull request #902 from cloudwego/release/v0.5.2

chore: release v0.5.2

.github/PULL_REQUEST_TEMPLATE.md

@@ -23,6 +23,7 @@ Please check your PR title with the below requirements:
 -->
 - [ ] This PR title match the format: \<type\>(optional scope): \<description\>
 - [ ] The description of this PR title is user-oriented and clear enough for others to understand.
+- [ ] Attach the PR updating the user documentation if the current PR requires user awareness at the usage level. [User docs repo](https://github.com/cloudwego/cloudwego.github.io)
 
 
 #### (Optional) Translate the PR title into Chinese.
@@ -35,8 +36,13 @@ Provide more detailed info for review(e.g., it's recommended to provide perf dat
 en: 
 zh(optional): 
 
-#### Which issue(s) this PR fixes:
+#### (Optional) Which issue(s) this PR fixes:
 <!--
 Automatically closes linked issue when PR is merged.
 Eg: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
 -->
+
+#### (optional) The PR that updates user documentation:
+<!--
+If the current PR requires user awareness at the usage level, please submit a PR to update user docs. [User docs repo](https://github.com/cloudwego/cloudwego.github.io)
+-->

.licenserc.yaml

@@ -32,5 +32,6 @@ header:
     - pkg/generic/descriptor/tree_test.go
     - pkg/generic/httppb_test/idl/echo.pb.go
     - pkg/utils/json.go
+    - pkg/protocol/bthrift/test/kitex_gen/**
 
   comment: on-failure

client/client_test.go

@@ -544,15 +544,24 @@ func TestRetryWithResultRetry(t *testing.T) {
 
 	mockErr := errors.New("mock")
 	retryWithMockErr := false
-	var count int32
 	errMW := func(next endpoint.Endpoint) endpoint.Endpoint {
+		var count int32
 		return func(ctx context.Context, req, resp interface{}) (err error) {
 			if atomic.CompareAndSwapInt32(&count, 0, 1) {
 				return mockErr
 			}
 			return nil
 		}
 	}
+	mockTimeoutMW := func(next endpoint.Endpoint) endpoint.Endpoint {
+		var count int32
+		return func(ctx context.Context, req, resp interface{}) (err error) {
+			if atomic.CompareAndSwapInt32(&count, 0, 1) {
+				time.Sleep(300 * time.Millisecond)
+			}
+			return nil
+		}
+	}
 	errRetryFunc := func(err error, ri rpcinfo.RPCInfo) bool {
 		if errors.Is(err, mockErr) {
 			retryWithMockErr = true
@@ -561,7 +570,7 @@ func TestRetryWithResultRetry(t *testing.T) {
 		return false
 	}
 
-	// should timeout
+	// case 1: retry for mockErr
 	cli := newMockClient(t, ctrl,
 		WithMiddleware(errMW),
 		WithRPCTimeout(100*time.Millisecond),
@@ -580,6 +589,40 @@ func TestRetryWithResultRetry(t *testing.T) {
 	err := cli.Call(context.Background(), mtd, req, res)
 	test.Assert(t, err == nil, err)
 	test.Assert(t, retryWithMockErr)
+
+	// case 1: retry for timeout
+	cli = newMockClient(t, ctrl,
+		WithMiddleware(mockTimeoutMW),
+		WithRPCTimeout(100*time.Millisecond),
+		WithFailureRetry(&retry.FailurePolicy{
+			StopPolicy: retry.StopPolicy{
+				MaxRetryTimes: 3,
+				CBPolicy: retry.CBPolicy{
+					ErrorRate: 0.1,
+				},
+			},
+			RetrySameNode:     true,
+			ShouldResultRetry: &retry.ShouldResultRetry{ErrorRetry: errRetryFunc},
+		}))
+	err = cli.Call(context.Background(), mtd, req, res)
+	test.Assert(t, err == nil, err)
+
+	// case 2: set NotRetryForTimeout as true, won't do retry for timeout
+	cli = newMockClient(t, ctrl,
+		WithMiddleware(mockTimeoutMW),
+		WithRPCTimeout(100*time.Millisecond),
+		WithFailureRetry(&retry.FailurePolicy{
+			StopPolicy: retry.StopPolicy{
+				MaxRetryTimes: 3,
+				CBPolicy: retry.CBPolicy{
+					ErrorRate: 0.1,
+				},
+			},
+			RetrySameNode:     true,
+			ShouldResultRetry: &retry.ShouldResultRetry{ErrorRetry: errRetryFunc, NotRetryForTimeout: true},
+		}))
+	err = cli.Call(context.Background(), mtd, req, res)
+	test.Assert(t, errors.Is(err, kerrors.ErrRPCTimeout))
 }
 
 func TestFallbackForError(t *testing.T) {

client/rpctimeout.go

@@ -61,9 +61,14 @@ func makeTimeoutErr(ctx context.Context, start time.Time, timeout time.Duration)
 		errMsg = fmt.Sprintf("%s, remote=%s", errMsg, target.String())
 	}
 
+	needFineGrainedErrCode := rpctimeout.LoadGlobalNeedFineGrainedErrCode()
 	// cancel error
 	if ctx.Err() == context.Canceled {
-		return kerrors.ErrRPCTimeout.WithCause(fmt.Errorf("%s: %w by business", errMsg, ctx.Err()))
+		if needFineGrainedErrCode {
+			return kerrors.ErrCanceledByBusiness
+		} else {
+			return kerrors.ErrRPCTimeout.WithCause(fmt.Errorf("%s: %w by business", errMsg, ctx.Err()))
+		}
 	}
 
 	if ddl, ok := ctx.Deadline(); !ok {
@@ -75,10 +80,19 @@ func makeTimeoutErr(ctx context.Context, start time.Time, timeout time.Duration)
 		if roundTimeout >= 0 && ddl.Before(start.Add(roundTimeout)) {
 			errMsg = fmt.Sprintf("%s, context deadline earlier than timeout, actual=%v", errMsg, ddl.Sub(start))
 		}
+
+		if needFineGrainedErrCode && isBusinessTimeout(start, timeout, ddl, rpctimeout.LoadBusinessTimeoutThreshold()) {
+			return kerrors.ErrTimeoutByBusiness
+		}
 	}
 	return kerrors.ErrRPCTimeout.WithCause(errors.New(errMsg))
 }
 
+func isBusinessTimeout(start time.Time, kitexTimeout time.Duration, actualDDL time.Time, threshold time.Duration) bool {
+	kitexDDL := start.Add(kitexTimeout)
+	return actualDDL.Add(threshold).Before(kitexDDL)
+}
+
 func rpcTimeoutMW(mwCtx context.Context) endpoint.Middleware {
 	var moreTimeout time.Duration
 	if v, ok := mwCtx.Value(rpctimeout.TimeoutAdjustKey).(*time.Duration); ok && v != nil {

client/rpctimeout_test.go

@@ -99,3 +99,121 @@ func TestNewRPCTimeoutMW(t *testing.T) {
 	err = mw1(mw2(block))(cancelCtx, nil, nil)
 	test.Assert(t, errors.Is(err, context.Canceled), err)
 }
+
+func TestIsBusinessTimeout(t *testing.T) {
+	type args struct {
+		start        time.Time
+		kitexTimeout time.Duration
+		actualDDL    time.Time
+		threshold    time.Duration
+	}
+	start := time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
+	tests := []struct {
+		name string
+		args args
+		want bool
+	}{
+		{
+			name: "business ddl ahead of kitex ddl by more than threshold",
+			args: args{
+				start:        start,
+				kitexTimeout: time.Second * 2,
+				actualDDL:    start.Add(time.Second),
+				threshold:    time.Millisecond * 500,
+			},
+			want: true,
+		},
+		{
+			name: "business ddl ahead of kitex ddl by less than threshold",
+			args: args{
+				start:        start,
+				kitexTimeout: time.Second,
+				actualDDL:    start.Add(time.Millisecond * 800),
+				threshold:    time.Millisecond * 500,
+			},
+			want: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := isBusinessTimeout(tt.args.start, tt.args.kitexTimeout, tt.args.actualDDL, tt.args.threshold); got != tt.want {
+				t.Errorf("isBusinessTimeout() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestRpcTimeoutMWTimeoutByBusiness(t *testing.T) {
+	runTimeoutMW := func() error {
+		mw := rpcTimeoutMW(context.Background())
+		processor := mw(func(ctx context.Context, req, rsp interface{}) error {
+			time.Sleep(time.Millisecond * 100)
+			return nil
+		})
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*30)
+		defer cancel()
+
+		ctx = rpcinfo.NewCtxWithRPCInfo(ctx, mockRPCInfo(time.Second))
+
+		return processor(ctx, nil, nil)
+	}
+
+	t.Run("Timeout by business with no need fine grained err code", func(t *testing.T) {
+		rpctimeout.DisableGlobalNeedFineGrainedErrCode()
+		if err := runTimeoutMW(); !kerrors.IsTimeoutError(err) {
+			t.Errorf("rpcTimeoutMW() = %v, want %v", err, kerrors.ErrRPCTimeout)
+		}
+	})
+
+	t.Run("Timeout by business with need fine grained err code", func(t *testing.T) {
+		rpctimeout.EnableGlobalNeedFineGrainedErrCode()
+		if err := runTimeoutMW(); err != kerrors.ErrTimeoutByBusiness {
+			t.Errorf("rpcTimeoutMW() = %v, want %v", err, kerrors.ErrTimeoutByBusiness)
+		}
+		rpctimeout.DisableGlobalNeedFineGrainedErrCode()
+	})
+}
+
+func TestRpcTimeoutMWCancelByBusiness(t *testing.T) {
+	runTimeoutMW := func() error {
+		mw := rpcTimeoutMW(context.Background())
+		processor := mw(func(ctx context.Context, req, rsp interface{}) error {
+			time.Sleep(time.Millisecond * 100)
+			return nil
+		})
+
+		ctx, cancel := context.WithCancel(context.Background())
+		go func() {
+			time.Sleep(10 * time.Millisecond)
+			cancel()
+		}()
+
+		ctx = rpcinfo.NewCtxWithRPCInfo(ctx, mockRPCInfo(time.Second))
+
+		return processor(ctx, nil, nil)
+	}
+
+	t.Run("Cancel by business with no need fine grained err code", func(t *testing.T) {
+		rpctimeout.DisableGlobalNeedFineGrainedErrCode()
+		if err := runTimeoutMW(); !kerrors.IsTimeoutError(err) {
+			t.Errorf("rpcTimeoutMW() = %v, want %v", err, kerrors.ErrRPCTimeout)
+		}
+	})
+
+	t.Run("Cancel by business with need fine grained err code", func(t *testing.T) {
+		rpctimeout.EnableGlobalNeedFineGrainedErrCode()
+		if err := runTimeoutMW(); err != kerrors.ErrCanceledByBusiness {
+			t.Errorf("rpcTimeoutMW() = %v, want %v", err, kerrors.ErrCanceledByBusiness)
+		}
+		rpctimeout.DisableGlobalNeedFineGrainedErrCode()
+	})
+}
+
+func mockRPCInfo(timeout time.Duration) rpcinfo.RPCInfo {
+	s := rpcinfo.NewEndpointInfo("mockService", "mockMethod", nil, nil)
+	c := rpcinfo.NewRPCConfig()
+	mc := rpcinfo.AsMutableRPCConfig(c)
+	_ = mc.SetRPCTimeout(timeout)
+	return rpcinfo.NewRPCInfo(nil, s, nil, c, rpcinfo.NewRPCStats())
+}

client/stream_test.go

@@ -102,13 +102,24 @@ func TestStreaming(t *testing.T) {
 		kc:     kc,
 	}
 
-	// recv nil msg
-	err = stream.RecvMsg(nil)
-	test.Assert(t, err == nil, err)
-
-	// send nil msg
-	err = stream.SendMsg(nil)
-	test.Assert(t, err == nil, err)
+	for i := 0; i < 10; i++ {
+		ch := make(chan struct{})
+		// recv nil msg
+		go func() {
+			err := stream.RecvMsg(nil)
+			test.Assert(t, err == nil, err)
+			ch <- struct{}{}
+		}()
+
+		// send nil msg
+		go func() {
+			err := stream.SendMsg(nil)
+			test.Assert(t, err == nil, err)
+			ch <- struct{}{}
+		}()
+		<-ch
+		<-ch
+	}
 
 	// close
 	err = stream.Close()

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/cloudwego/fastpb v0.0.4
 	github.com/cloudwego/frugal v0.1.6
 	github.com/cloudwego/netpoll v0.3.2
-	github.com/cloudwego/thriftgo v0.2.8
+	github.com/cloudwego/thriftgo v0.2.9
 	github.com/golang/mock v1.6.0
 	github.com/google/pprof v0.0.0-20220608213341-c488b8fa1db3
 	github.com/jhump/protoreflect v1.8.2

go.sum

@@ -26,16 +26,14 @@ github.com/chzyer/logex v1.2.0/go.mod h1:9+9sk7u7pGNWYMkh0hdiL++6OeibzJccyQU4p4M
 github.com/chzyer/readline v1.5.0/go.mod h1:x22KAscuvRqlLoK9CsoYsmxoXZMMFVyOl86cAH8qUic=
 github.com/chzyer/test v0.0.0-20210722231415-061457976a23/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
-github.com/cloudwego/fastpb v0.0.4-0.20230131074846-6fc453d58b96 h1:61PQT0CXNUuQDiDKv/QQ+pFi9uthExZLQz8b5WfS7Qw=
-github.com/cloudwego/fastpb v0.0.4-0.20230131074846-6fc453d58b96/go.mod h1:/V13XFTq2TUkxj2qWReV8MwfPC4NnPcy6FsrojnsSG0=
 github.com/cloudwego/fastpb v0.0.4 h1:/ROVVfoFtpfc+1pkQLzGs+azjxUbSOsAqSY4tAAx4mg=
 github.com/cloudwego/fastpb v0.0.4/go.mod h1:/V13XFTq2TUkxj2qWReV8MwfPC4NnPcy6FsrojnsSG0=
 github.com/cloudwego/frugal v0.1.6 h1:aXJ7W0Omion1WTCe4JHAWinQmjXDYzHt03sabu3Rabo=
 github.com/cloudwego/frugal v0.1.6/go.mod h1:9ElktKsh5qd2zDBQ5ENhPSQV7F2dZ/mXlr1eaZGDBFs=
 github.com/cloudwego/netpoll v0.3.2 h1:/998ICrNMVBo4mlul4j7qcIeY7QnEfuCCPPwck9S3X4=
 github.com/cloudwego/netpoll v0.3.2/go.mod h1:xVefXptcyheopwNDZjDPcfU6kIjZXZ4nY550k1yH9eQ=
-github.com/cloudwego/thriftgo v0.2.8 h1:swwp+JQDeL8bBbvzJN3D3J5fluWP+chiUqVPbnToV0I=
-github.com/cloudwego/thriftgo v0.2.8/go.mod h1:dAyXHEmKXo0LfMCrblVEY3mUZsdeuA5+i0vF5f09j7E=
+github.com/cloudwego/thriftgo v0.2.9 h1:uN58Y6LkVXHWWEFcDFVi4gAnvmFTxTqgi26NZaRAubQ=
+github.com/cloudwego/thriftgo v0.2.9/go.mod h1:dAyXHEmKXo0LfMCrblVEY3mUZsdeuA5+i0vF5f09j7E=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=

pkg/kerrors/kerrors.go

@@ -26,18 +26,20 @@ import (
 
 // Basic error types
 var (
-	ErrInternalException = &basicError{"internal exception"}
-	ErrServiceDiscovery  = &basicError{"service discovery error"}
-	ErrGetConnection     = &basicError{"get connection error"}
-	ErrLoadbalance       = &basicError{"loadbalance error"}
-	ErrNoMoreInstance    = &basicError{"no more instances to retry"}
-	ErrRPCTimeout        = &basicError{"rpc timeout"}
-	ErrACL               = &basicError{"request forbidden"}
-	ErrCircuitBreak      = &basicError{"forbidden by circuitbreaker"}
-	ErrRemoteOrNetwork   = &basicError{"remote or network error"}
-	ErrOverlimit         = &basicError{"request over limit"}
-	ErrPanic             = &basicError{"panic"}
-	ErrBiz               = &basicError{"biz error"}
+	ErrInternalException  = &basicError{"internal exception"}
+	ErrServiceDiscovery   = &basicError{"service discovery error"}
+	ErrGetConnection      = &basicError{"get connection error"}
+	ErrLoadbalance        = &basicError{"loadbalance error"}
+	ErrNoMoreInstance     = &basicError{"no more instances to retry"}
+	ErrRPCTimeout         = &basicError{"rpc timeout"}
+	ErrCanceledByBusiness = &basicError{"canceled by business"}
+	ErrTimeoutByBusiness  = &basicError{"timeout by business"}
+	ErrACL                = &basicError{"request forbidden"}
+	ErrCircuitBreak       = &basicError{"forbidden by circuitbreaker"}
+	ErrRemoteOrNetwork    = &basicError{"remote or network error"}
+	ErrOverlimit          = &basicError{"request over limit"}
+	ErrPanic              = &basicError{"panic"}
+	ErrBiz                = &basicError{"biz error"}
 
 	ErrRetry = &basicError{"retry error"}
 	// ErrRPCFinish happens when retry enabled and there is one call has finished

pkg/protocol/bthrift/test/gen.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+kitex -module github.com/cloudwego/kitex -thrift keep_unknown_fields test.thrift 
+

pkg/protocol/bthrift/test/kitex_gen/test/k-consts.go

@@ -0,0 +1,4 @@
+package test
+
+// KitexUnusedProtection is used to prevent 'imported and not used' error.
+var KitexUnusedProtection = struct{}{}