From d19fdb7bd2fe7c42ed8c2ba1d509a994c57c7c6d Mon Sep 17 00:00:00 2001 From: Rupayon Haldar <80724680+rupayon123@users.noreply.github.com> Date: Fri, 11 Sep 2026 07:45:48 -0400 Subject: [PATCH 1/4] fix(grpc): return non-OK RPC status to the caller --- internal/igrpc/grpc.go | 1 + internal/igrpc/status_test.go | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+) create mode 100644 internal/igrpc/status_test.go diff --git a/internal/igrpc/grpc.go b/internal/igrpc/grpc.go index 640e06c..fd649fe 100644 --- a/internal/igrpc/grpc.go +++ b/internal/igrpc/grpc.go @@ -125,6 +125,7 @@ func Run(ctx context.Context, opts Options) error { if h.Status.Code() != 0 { grpcurl.PrintStatus(os.Stderr, h.Status, formatter) + return h.Status.Err() } return nil diff --git a/internal/igrpc/status_test.go b/internal/igrpc/status_test.go new file mode 100644 index 0000000..78fcaf0 --- /dev/null +++ b/internal/igrpc/status_test.go @@ -0,0 +1,35 @@ +package igrpc_test + +import ( + "context" + "net" + "testing" + "time" + + "github.com/kavix/kurl/internal/igrpc" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/reflection" + "google.golang.org/grpc/status" +) + +func TestRunReturnsRPCStatus(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer() + grpc_health_v1.RegisterHealthServer(server, health.NewServer()) + reflection.Register(server) + go func() { _ = server.Serve(listener) }() + defer server.Stop() + err = igrpc.Run(context.Background(), igrpc.Options{ + URL: "grpc://" + listener.Addr().String(), Method: "grpc.health.v1.Health/Check", + Data: `{"service":"missing"}`, Timeout: 5 * time.Second, + }) + if status.Code(err) != codes.NotFound { + t.Fatalf("got %v (%v), want NotFound", err, status.Code(err)) + } +} From 2cd9c499b666e5bf78c8da89506fa8c101886b18 Mon Sep 17 00:00:00 2001 From: Rupayon Haldar <80724680+rupayon123@users.noreply.github.com> Date: Fri, 11 Sep 2026 07:46:42 -0400 Subject: [PATCH 2/4] fix(grpc): apply configured timeout to RPC invocation --- internal/igrpc/grpc.go | 2 +- internal/igrpc/timeout_test.go | 61 ++++++++++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 1 deletion(-) create mode 100644 internal/igrpc/timeout_test.go diff --git a/internal/igrpc/grpc.go b/internal/igrpc/grpc.go index fd649fe..21d2d1f 100644 --- a/internal/igrpc/grpc.go +++ b/internal/igrpc/grpc.go @@ -118,7 +118,7 @@ func Run(ctx context.Context, opts Options) error { h.VerbosityLevel = 1 } - err = grpcurl.InvokeRPC(ctx, descSource, cc, opts.Method, opts.Headers, h, rf.Next) + err = grpcurl.InvokeRPC(dialCtx, descSource, cc, opts.Method, opts.Headers, h, rf.Next) if err != nil { return fmt.Errorf("rpc error: %w", err) } diff --git a/internal/igrpc/timeout_test.go b/internal/igrpc/timeout_test.go new file mode 100644 index 0000000..41e7ea2 --- /dev/null +++ b/internal/igrpc/timeout_test.go @@ -0,0 +1,61 @@ +package igrpc_test + +import ( + "context" + "net" + "testing" + "time" + + "github.com/kavix/kurl/internal/igrpc" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/reflection" + "google.golang.org/grpc/status" +) + +type waitingHealthServer struct { + grpc_health_v1.UnimplementedHealthServer + remaining chan time.Duration +} + +func (s *waitingHealthServer) Check(ctx context.Context, _ *grpc_health_v1.HealthCheckRequest) (*grpc_health_v1.HealthCheckResponse, error) { + deadline, ok := ctx.Deadline() + if !ok { + s.remaining <- time.Hour + } else { + s.remaining <- time.Until(deadline) + } + <-ctx.Done() + return nil, status.FromContextError(ctx.Err()).Err() +} + +func TestRunAppliesTimeoutToInvocation(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer() + healthServer := &waitingHealthServer{remaining: make(chan time.Duration, 1)} + grpc_health_v1.RegisterHealthServer(server, healthServer) + reflection.Register(server) + go func() { _ = server.Serve(listener) }() + defer server.Stop() + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + err = igrpc.Run(ctx, igrpc.Options{ + URL: "grpc://" + listener.Addr().String(), Method: "grpc.health.v1.Health/Check", + Data: `{}`, Timeout: time.Second, + }) + if status.Code(err) != codes.DeadlineExceeded { + t.Fatalf("got %v, want deadline exceeded", err) + } + select { + case remaining := <-healthServer.remaining: + if remaining > time.Second { + t.Errorf("RPC received a %s deadline, exceeding the configured one-second timeout", remaining) + } + default: + t.Fatal("RPC did not reach the server") + } +} From f0f7ab22d8ba37a623b0cdd0a79c9bd0331a6518 Mon Sep 17 00:00:00 2001 From: Rupayon Haldar <80724680+rupayon123@users.noreply.github.com> Date: Fri, 11 Sep 2026 07:47:55 -0400 Subject: [PATCH 3/4] fix(grpc): include request headers in server reflection --- internal/igrpc/grpc.go | 6 ++- internal/igrpc/reflection_headers_test.go | 56 +++++++++++++++++++++++ 2 files changed, 60 insertions(+), 2 deletions(-) create mode 100644 internal/igrpc/reflection_headers_test.go diff --git a/internal/igrpc/grpc.go b/internal/igrpc/grpc.go index 21d2d1f..0650a1d 100644 --- a/internal/igrpc/grpc.go +++ b/internal/igrpc/grpc.go @@ -13,6 +13,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/credentials" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/metadata" ) type Options struct { @@ -72,9 +73,10 @@ func Run(ctx context.Context, opts Options) error { return fmt.Errorf("failed to process proto file: %w", err) } } else { - refClient := grpcreflect.NewClientAuto(dialCtx, cc) + reflectionCtx := metadata.NewOutgoingContext(dialCtx, grpcurl.MetadataFromHeaders(opts.Headers)) + refClient := grpcreflect.NewClientAuto(reflectionCtx, cc) defer refClient.Reset() - descSource = grpcurl.DescriptorSourceFromServer(dialCtx, refClient) + descSource = grpcurl.DescriptorSourceFromServer(reflectionCtx, refClient) } if opts.ListServices { diff --git a/internal/igrpc/reflection_headers_test.go b/internal/igrpc/reflection_headers_test.go new file mode 100644 index 0000000..bb220ec --- /dev/null +++ b/internal/igrpc/reflection_headers_test.go @@ -0,0 +1,56 @@ +package igrpc_test + +import ( + "context" + "net" + "testing" + "time" + + "github.com/kavix/kurl/internal/igrpc" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/reflection" + "google.golang.org/grpc/status" +) + +func TestRunSendsHeadersToReflection(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + authenticated := func(ctx context.Context) bool { + values := metadata.ValueFromIncomingContext(ctx, "authorization") + return len(values) == 1 && values[0] == "Bearer test-token" + } + server := grpc.NewServer( + grpc.StreamInterceptor(func(srv interface{}, stream grpc.ServerStream, info *grpc.StreamServerInfo, handler grpc.StreamHandler) error { + if !authenticated(stream.Context()) { + return status.Error(codes.Unauthenticated, "reflection requires authorization") + } + return handler(srv, stream) + }), + grpc.UnaryInterceptor(func(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) { + if !authenticated(ctx) { + return nil, status.Error(codes.Unauthenticated, "RPC requires authorization") + } + return handler(ctx, req) + }), + ) + grpc_health_v1.RegisterHealthServer(server, health.NewServer()) + reflection.Register(server) + go func() { _ = server.Serve(listener) }() + defer server.Stop() + for _, list := range []bool{true, false} { + err := igrpc.Run(context.Background(), igrpc.Options{ + URL: "grpc://" + listener.Addr().String(), ListServices: list, + Method: "grpc.health.v1.Health/Check", Data: `{}`, + Headers: []string{"Authorization: Bearer test-token"}, Timeout: 5 * time.Second, + }) + if err != nil { + t.Errorf("list=%v: %v", list, err) + } + } +} From 3b2902fd97a0b89ada5f9a45d1e211e324c4c719 Mon Sep 17 00:00:00 2001 From: Rupayon Haldar <80724680+rupayon123@users.noreply.github.com> Date: Fri, 11 Sep 2026 13:20:44 -0400 Subject: [PATCH 4/4] fix(grpc): allow zero timeout to disable the deadline --- internal/igrpc/disabled_timeout_test.go | 31 +++++++++++++++++++++++++ internal/igrpc/grpc.go | 8 +++++-- 2 files changed, 37 insertions(+), 2 deletions(-) create mode 100644 internal/igrpc/disabled_timeout_test.go diff --git a/internal/igrpc/disabled_timeout_test.go b/internal/igrpc/disabled_timeout_test.go new file mode 100644 index 0000000..b8a3452 --- /dev/null +++ b/internal/igrpc/disabled_timeout_test.go @@ -0,0 +1,31 @@ +package igrpc_test + +import ( + "context" + "github.com/kavix/kurl/internal/igrpc" + "google.golang.org/grpc" + "google.golang.org/grpc/health" + "google.golang.org/grpc/health/grpc_health_v1" + "google.golang.org/grpc/reflection" + "net" + "testing" + "time" +) + +func TestRunAllowsDisabledTimeout(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer() + grpc_health_v1.RegisterHealthServer(server, health.NewServer()) + reflection.Register(server) + go func() { _ = server.Serve(listener) }() + defer server.Stop() + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + err = igrpc.Run(ctx, igrpc.Options{URL: "grpc://" + listener.Addr().String(), Method: "grpc.health.v1.Health/Check", Data: `{}`, Timeout: 0}) + if err != nil { + t.Fatalf("disabled timeout prevented RPC: %v", err) + } +} diff --git a/internal/igrpc/grpc.go b/internal/igrpc/grpc.go index 0650a1d..8a8b56c 100644 --- a/internal/igrpc/grpc.go +++ b/internal/igrpc/grpc.go @@ -56,8 +56,12 @@ func Run(ctx context.Context, opts Options) error { creds = insecure.NewCredentials() } - dialCtx, cancel := context.WithTimeout(ctx, opts.Timeout) - defer cancel() + dialCtx := ctx + if opts.Timeout > 0 { + var cancel context.CancelFunc + dialCtx, cancel = context.WithTimeout(ctx, opts.Timeout) + defer cancel() + } cc, err := grpc.DialContext(dialCtx, target, grpc.WithTransportCredentials(creds)) if err != nil {