Skip to content

Commit 8bbc0d8

Browse files
ChocoLZSChocoLZS
committed
add unit test for rpcinfo async usage
1 parent b37b9bd commit 8bbc0d8

10 files changed

Lines changed: 895 additions & 23 deletions

File tree

‎client/client_test.go‎

Lines changed: 181 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ import (
2323
"runtime"
2424
"runtime/debug"
2525
"strings"
26+
"sync"
2627
"sync/atomic"
2728
"testing"
2829
"time"
@@ -152,6 +153,186 @@ func TestCall(t *testing.T) {
152153
test.Assert(t, err == nil, err)
153154
}
154155

156+
type asyncRPCInfoReadTracker struct {
157+
mu sync.Mutex
158+
stops []chan struct{}
159+
dones []chan any
160+
}
161+
162+
func (r *asyncRPCInfoReadTracker) middleware(next endpoint.Endpoint) endpoint.Endpoint {
163+
return func(ctx context.Context, req, resp interface{}) error {
164+
stop := make(chan struct{})
165+
started := make(chan struct{})
166+
done := make(chan any, 1)
167+
168+
r.mu.Lock()
169+
r.stops = append(r.stops, stop)
170+
r.dones = append(r.dones, done)
171+
r.mu.Unlock()
172+
173+
go readRPCInfoUntilStopped(ctx, stop, started, done)
174+
<-started
175+
return next(ctx, req, resp)
176+
}
177+
}
178+
179+
func (r *asyncRPCInfoReadTracker) stopAndAssert(t *testing.T) {
180+
t.Helper()
181+
182+
r.mu.Lock()
183+
stops := append([]chan struct{}(nil), r.stops...)
184+
dones := append([]chan any(nil), r.dones...)
185+
r.mu.Unlock()
186+
187+
for _, stop := range stops {
188+
close(stop)
189+
}
190+
for _, done := range dones {
191+
if panicInfo := <-done; panicInfo != nil {
192+
t.Fatalf("async RPCInfo read panicked: %v", panicInfo)
193+
}
194+
}
195+
}
196+
197+
func TestCallDisablePoolKeepsRPCInfoReadableAcrossLifecycle(t *testing.T) {
198+
mockErr := errors.New("mock")
199+
testcases := []struct {
200+
name string
201+
options func(*asyncRPCInfoReadTracker) []Option
202+
assertErr func(error)
203+
}{
204+
{
205+
name: "no retry success recycles",
206+
options: func(reader *asyncRPCInfoReadTracker) []Option {
207+
return []Option{WithMiddleware(reader.middleware)}
208+
},
209+
assertErr: func(err error) {
210+
test.Assert(t, err == nil, err)
211+
},
212+
},
213+
{
214+
name: "no retry error skips recycle",
215+
options: func(reader *asyncRPCInfoReadTracker) []Option {
216+
errMW := func(next endpoint.Endpoint) endpoint.Endpoint {
217+
return func(ctx context.Context, req, resp interface{}) error {
218+
return mockErr
219+
}
220+
}
221+
return []Option{WithMiddleware(reader.middleware), WithMiddleware(errMW)}
222+
},
223+
assertErr: func(err error) {
224+
test.Assert(t, errors.Is(err, mockErr), err)
225+
},
226+
},
227+
{
228+
name: "failure retry configured without retry recycles",
229+
options: func(reader *asyncRPCInfoReadTracker) []Option {
230+
return []Option{
231+
WithMiddleware(reader.middleware),
232+
WithFailureRetry(&retry.FailurePolicy{
233+
StopPolicy: retry.StopPolicy{
234+
MaxRetryTimes: 1,
235+
CBPolicy: retry.CBPolicy{ErrorRate: 0.1},
236+
},
237+
}),
238+
}
239+
},
240+
assertErr: func(err error) {
241+
test.Assert(t, err == nil, err)
242+
},
243+
},
244+
{
245+
name: "failure retry actual retry skips recycle",
246+
options: func(reader *asyncRPCInfoReadTracker) []Option {
247+
var callTimes int32
248+
errMW := func(next endpoint.Endpoint) endpoint.Endpoint {
249+
return func(ctx context.Context, req, resp interface{}) error {
250+
if atomic.AddInt32(&callTimes, 1) == 1 {
251+
return mockErr
252+
}
253+
return next(ctx, req, resp)
254+
}
255+
}
256+
return []Option{
257+
WithMiddleware(reader.middleware),
258+
WithMiddleware(errMW),
259+
WithFailureRetry(&retry.FailurePolicy{
260+
StopPolicy: retry.StopPolicy{
261+
MaxRetryTimes: 1,
262+
CBPolicy: retry.CBPolicy{ErrorRate: 0.1},
263+
},
264+
ShouldResultRetry: &retry.ShouldResultRetry{
265+
ErrorRetry: func(err error, ri rpcinfo.RPCInfo) bool {
266+
return errors.Is(err, mockErr)
267+
},
268+
},
269+
}),
270+
}
271+
},
272+
assertErr: func(err error) {
273+
test.Assert(t, err == nil, err)
274+
},
275+
},
276+
{
277+
name: "backup retry skips recycle",
278+
options: func(reader *asyncRPCInfoReadTracker) []Option {
279+
var callTimes int32
280+
slowFirstCallMW := func(next endpoint.Endpoint) endpoint.Endpoint {
281+
return func(ctx context.Context, req, resp interface{}) error {
282+
if atomic.AddInt32(&callTimes, 1) == 1 {
283+
time.Sleep(20 * time.Millisecond)
284+
}
285+
return next(ctx, req, resp)
286+
}
287+
}
288+
return []Option{
289+
WithMiddleware(reader.middleware),
290+
WithMiddleware(slowFirstCallMW),
291+
WithBackupRequest(retry.NewBackupPolicy(1)),
292+
}
293+
},
294+
assertErr: func(err error) {
295+
test.Assert(t, err == nil, err)
296+
},
297+
},
298+
{
299+
name: "mixed retry skips recycle",
300+
options: func(reader *asyncRPCInfoReadTracker) []Option {
301+
var callTimes int32
302+
slowFirstCallMW := func(next endpoint.Endpoint) endpoint.Endpoint {
303+
return func(ctx context.Context, req, resp interface{}) error {
304+
if atomic.AddInt32(&callTimes, 1) == 1 {
305+
time.Sleep(20 * time.Millisecond)
306+
}
307+
return next(ctx, req, resp)
308+
}
309+
}
310+
return []Option{
311+
WithMiddleware(reader.middleware),
312+
WithMiddleware(slowFirstCallMW),
313+
WithMixedRetry(retry.NewMixedPolicy(1)),
314+
}
315+
},
316+
assertErr: func(err error) {
317+
test.Assert(t, err == nil, err)
318+
},
319+
},
320+
}
321+
322+
for _, tc := range testcases {
323+
t.Run(tc.name, func(t *testing.T) {
324+
ctrl := gomock.NewController(t)
325+
defer ctrl.Finish()
326+
327+
reader := &asyncRPCInfoReadTracker{}
328+
cli := newMockClient(t, ctrl, tc.options(reader)...)
329+
err := cli.Call(context.Background(), mocks.MockMethod, mocks.NewMockArgs(), mocks.NewMockResult())
330+
tc.assertErr(err)
331+
reader.stopAndAssert(t)
332+
})
333+
}
334+
}
335+
155336
func TestCallWithContextBackup(t *testing.T) {
156337
localsession.InitDefaultManager(localsession.DefaultManagerOptions())
157338
d, dd := "d", "dd"

‎client/rpctimeout_test.go‎

Lines changed: 122 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package client
1919
import (
2020
"context"
2121
"errors"
22+
"runtime"
2223
"strings"
2324
"testing"
2425
"time"
@@ -46,6 +47,79 @@ func panicEp(ctx context.Context, request, response interface{}) (err error) {
4647
panic(panicMsg)
4748
}
4849

50+
func mustReadRPCInfoAsync(t *testing.T, ctx context.Context) {
51+
t.Helper()
52+
53+
done := make(chan any, 1)
54+
go func() {
55+
defer func() {
56+
done <- recover()
57+
}()
58+
readRPCInfoForAsyncTest(ctx)
59+
}()
60+
if panicInfo := <-done; panicInfo != nil {
61+
t.Fatalf("async RPCInfo read panicked: %v", panicInfo)
62+
}
63+
}
64+
65+
// readRPCInfoForAsyncTest touches RPCInfo fields commonly reset by framework
66+
// recycle paths. Normal test runs catch nil/reset panics; -race catches hidden
67+
// framework writes racing with an asynchronous user reader.
68+
func readRPCInfoForAsyncTest(ctx context.Context) {
69+
ri := rpcinfo.GetRPCInfo(ctx)
70+
if ri == nil {
71+
panic("nil RPCInfo")
72+
}
73+
if from := ri.From(); from == nil {
74+
panic("nil From endpoint")
75+
} else {
76+
_ = from.ServiceName()
77+
_ = from.Method()
78+
_ = from.Address()
79+
}
80+
if to := ri.To(); to == nil {
81+
panic("nil To endpoint")
82+
} else {
83+
_ = to.ServiceName()
84+
_ = to.Method()
85+
_ = to.Address()
86+
}
87+
if inv := ri.Invocation(); inv == nil {
88+
panic("nil Invocation")
89+
} else {
90+
_ = inv.ServiceName()
91+
_ = inv.MethodName()
92+
_ = inv.StreamingMode()
93+
}
94+
if cfg := ri.Config(); cfg == nil {
95+
panic("nil RPCConfig")
96+
} else {
97+
_ = cfg.RPCTimeout()
98+
}
99+
if stats := ri.Stats(); stats == nil {
100+
panic("nil RPCStats")
101+
} else {
102+
_ = stats.Level()
103+
_ = stats.Error()
104+
}
105+
}
106+
107+
func readRPCInfoUntilStopped(ctx context.Context, stop <-chan struct{}, started chan<- struct{}, done chan<- any) {
108+
close(started)
109+
defer func() {
110+
done <- recover()
111+
}()
112+
for {
113+
select {
114+
case <-stop:
115+
return
116+
default:
117+
readRPCInfoForAsyncTest(ctx)
118+
runtime.Gosched()
119+
}
120+
}
121+
}
122+
49123
func TestNewRPCTimeoutMW(t *testing.T) {
50124
t.Parallel()
51125

@@ -114,6 +188,41 @@ func TestNewRPCTimeoutMW(t *testing.T) {
114188
test.Panic(t, func() { mw1(mw2(panicEp))(ctx, nil, nil) })
115189
}
116190

191+
func TestRpcTimeoutMWDisablePoolKeepsRPCInfoReadableAfterTimeout(t *testing.T) {
192+
timeoutCtxCh := make(chan context.Context, 1)
193+
stop := make(chan struct{})
194+
started := make(chan struct{})
195+
done := make(chan any, 1)
196+
mw := rpcTimeoutMW(context.Background())
197+
processor := mw(func(ctx context.Context, req, rsp interface{}) error {
198+
select {
199+
case timeoutCtxCh <- ctx:
200+
default:
201+
}
202+
go readRPCInfoUntilStopped(ctx, stop, started, done)
203+
<-started
204+
time.Sleep(80 * time.Millisecond)
205+
return nil
206+
})
207+
208+
ctx := rpcinfo.NewCtxWithRPCInfo(context.Background(), mockFullRPCInfo(20*time.Millisecond))
209+
err := processor(ctx, nil, nil)
210+
test.Assert(t, errors.Is(err, kerrors.ErrRPCTimeout), err)
211+
212+
var timeoutCtx context.Context
213+
select {
214+
case timeoutCtx = <-timeoutCtxCh:
215+
case <-time.After(time.Second):
216+
t.Fatal("rpcTimeoutMW did not execute the background endpoint")
217+
}
218+
mustReadRPCInfoAsync(t, timeoutCtx)
219+
220+
close(stop)
221+
if panicInfo := <-done; panicInfo != nil {
222+
t.Fatalf("async RPCInfo read panicked: %v", panicInfo)
223+
}
224+
}
225+
117226
func TestIsBusinessTimeout(t *testing.T) {
118227
type args struct {
119228
start time.Time
@@ -250,6 +359,19 @@ func mockRPCInfo(timeout time.Duration) rpcinfo.RPCInfo {
250359
return rpcinfo.NewRPCInfo(nil, s, nil, c, rpcinfo.NewRPCStats())
251360
}
252361

362+
func mockFullRPCInfo(timeout time.Duration) rpcinfo.RPCInfo {
363+
c := rpcinfo.NewRPCConfig()
364+
mc := rpcinfo.AsMutableRPCConfig(c)
365+
_ = mc.SetRPCTimeout(timeout)
366+
return rpcinfo.NewRPCInfo(
367+
rpcinfo.NewEndpointInfo("mockCaller", "mockCallerMethod", nil, nil),
368+
rpcinfo.NewEndpointInfo("mockService", "mockMethod", nil, nil),
369+
rpcinfo.NewInvocation("mockService", "mockMethod"),
370+
c,
371+
rpcinfo.NewRPCStats(),
372+
)
373+
}
374+
253375
func Test_isBusinessTimeout(t *testing.T) {
254376
type args struct {
255377
start time.Time

‎client/service_inline_test.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,25 @@ func TestServiceInlineCall(t *testing.T) {
8484
test.Assert(t, err == nil, err)
8585
}
8686

87+
func TestServiceInlineDisablePoolKeepsClientRPCInfoReadableAfterCall(t *testing.T) {
88+
ctrl := gomock.NewController(t)
89+
defer ctrl.Finish()
90+
91+
var captured context.Context
92+
md := func(next endpoint.Endpoint) endpoint.Endpoint {
93+
return func(ctx context.Context, req, res interface{}) error {
94+
captured = ctx
95+
return next(ctx, req, res)
96+
}
97+
}
98+
cli := newMockServiceInlineClient(t, ctrl, WithMiddleware(md))
99+
100+
err := cli.Call(context.Background(), mocks.MockMethod, new(MockTStruct), new(MockTStruct))
101+
test.Assert(t, err == nil, err)
102+
test.Assert(t, captured != nil)
103+
mustReadRPCInfoAsync(t, captured)
104+
}
105+
87106
func TestServiceInlineTagOptions(t *testing.T) {
88107
ctrl := gomock.NewController(t)
89108
defer ctrl.Finish()

0 commit comments

Comments
 (0)