Summary
transport/grpc/client.go has an asymmetry between unary and streaming client interceptors:
// dial() — timeout passed to unary, but NOT to streaming
ints := []grpc.UnaryClientInterceptor{
unaryClientInterceptor(options.middleware, options.timeout, options.filters), // ✅ timeout applied
}
sints := []grpc.StreamClientInterceptor{
streamClientInterceptor(options.streamMiddleware, options.filters), // ❌ timeout ignored
}The streamClientInterceptor signature does not accept a timeout parameter, and the configured WithTimeout value is silently dropped for all streaming RPCs. As a result, streaming calls run indefinitely — bounded only by any deadline already present on the incoming context, not by the client-configured timeout.
Impact
Any service using streaming RPCs (bidirectional, server-streaming, client-streaming) via a go-kratos gRPC client will NOT have the WithTimeout deadline applied. The default unary timeout (2000ms) is effectively ignored for streams.
This becomes critical when a gRPC proxy (e.g. Easegress, Envoy) sits between the client and backend. If the proxy has a connection pool degradation issue — causing it to hold the stream open waiting for a backend connection — the streaming call has no bounded timeout and will hang until the parent context expires or the process is restarted.
In contrast, the same call as a unary RPC would be cancelled after WithTimeout milliseconds.
Current behaviour
func streamClientInterceptor(ms []middleware.Middleware, filters []selector.NodeFilter) grpc.StreamClientInterceptor {
return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string,
streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
// No timeout applied — ctx used as-is
...
clientStream, err := streamer(ctx, desc, cc, method, opts...)
...
}
}Expected behaviour
Streaming calls should respect WithTimeout the same way unary calls do:
func streamClientInterceptor(ms []middleware.Middleware, timeout time.Duration, filters []selector.NodeFilter) grpc.StreamClientInterceptor {
return func(ctx context.Context, desc *grpc.StreamDesc, cc *grpc.ClientConn, method string,
streamer grpc.Streamer, opts ...grpc.CallOption) (grpc.ClientStream, error) {
if timeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, timeout)
defer cancel()
}
...
clientStream, err := streamer(ctx, desc, cc, method, opts...)
...
}
}And dial() updated to pass the timeout:
sints := []grpc.StreamClientInterceptor{
streamClientInterceptor(options.streamMiddleware, options.timeout, options.filters),
}Fix
A PR is available: #[to be linked]
go-kratos version
v2.9.1 (confirmed in transport/grpc/client.go). The gap exists from the initial streaming interceptor implementation.
Summary
transport/grpc/client.gohas an asymmetry between unary and streaming client interceptors:The
streamClientInterceptorsignature does not accept atimeoutparameter, and the configuredWithTimeoutvalue is silently dropped for all streaming RPCs. As a result, streaming calls run indefinitely — bounded only by any deadline already present on the incoming context, not by the client-configured timeout.Impact
Any service using streaming RPCs (bidirectional, server-streaming, client-streaming) via a go-kratos gRPC client will NOT have the
WithTimeoutdeadline applied. The default unary timeout (2000ms) is effectively ignored for streams.This becomes critical when a gRPC proxy (e.g. Easegress, Envoy) sits between the client and backend. If the proxy has a connection pool degradation issue — causing it to hold the stream open waiting for a backend connection — the streaming call has no bounded timeout and will hang until the parent context expires or the process is restarted.
In contrast, the same call as a unary RPC would be cancelled after
WithTimeoutmilliseconds.Current behaviour
Expected behaviour
Streaming calls should respect
WithTimeoutthe same way unary calls do:And
dial()updated to pass the timeout:Fix
A PR is available: #[to be linked]
go-kratos version
v2.9.1 (confirmed in
transport/grpc/client.go). The gap exists from the initial streaming interceptor implementation.