diff --git a/internal/transport/http2_server.go b/internal/transport/http2_server.go index be8ae9f9c54f..428a3eed8e85 100644 --- a/internal/transport/http2_server.go +++ b/internal/transport/http2_server.go @@ -599,7 +599,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade cancelUpdated := make(chan struct{}) timer := internal.TimeAfterFunc(timeout, func() { <-cancelUpdated - t.closeStream(s, true, http2.ErrCodeCancel, false) + t.closeStream(s, true, http2.ErrCodeCancel, false, status.New(codes.DeadlineExceeded, context.DeadlineExceeded.Error())) }) oldCancel := s.cancel s.cancel = func() { @@ -654,7 +654,7 @@ func (t *http2Server) HandleStreams(ctx context.Context, handle func(*ServerStre s := t.activeStreams[se.StreamID] t.mu.Unlock() if s != nil { - t.closeStream(s, true, se.Code, false) + t.closeStream(s, true, se.Code, false, statusFromHTTP2Error(se.Code, se.Error())) } else { t.controlBuf.put(&cleanupStream{ streamID: se.StreamID, @@ -799,12 +799,12 @@ func (t *http2Server) handleData(f *parsedDataFrame) { return } if s.getState() == streamReadDone { - t.closeStream(s, true, http2.ErrCodeStreamClosed, false) + t.closeStream(s, true, http2.ErrCodeStreamClosed, false, status.New(codes.Internal, "transport: received data after the stream was closed")) return } if size > 0 { if err := s.fc.onData(size); err != nil { - t.closeStream(s, true, http2.ErrCodeFlowControl, false) + t.closeStream(s, true, http2.ErrCodeFlowControl, false, statusFromHTTP2Error(http2.ErrCodeFlowControl, err.Error())) return } dataLen := f.data.Len() @@ -831,7 +831,8 @@ func (t *http2Server) handleData(f *parsedDataFrame) { func (t *http2Server) handleRSTStream(f *http2.RSTStreamFrame) { // If the stream is not deleted from the transport's active streams map, then do a regular close stream. if s, ok := t.getStream(f); ok { - t.closeStream(s, false, 0, false) + st := statusFromHTTP2Error(f.ErrCode, fmt.Sprintf("stream terminated by RST_STREAM with error code: %v", f.ErrCode)) + t.closeStream(s, false, 0, false, st) return } // If the stream is already deleted from the active streams map, then put a cleanupStream item into controlbuf to delete the stream from loopy writer's established streams map. @@ -1061,7 +1062,7 @@ func (t *http2Server) writeHeaderLocked(s *ServerStream) error { if err != nil { return err } - t.closeStream(s, true, http2.ErrCodeInternal, false) + t.closeStream(s, true, http2.ErrCodeInternal, false, status.Convert(ErrHeaderListSizeLimitViolation)) return ErrHeaderListSizeLimitViolation } if t.stats != nil { @@ -1132,7 +1133,7 @@ func (t *http2Server) writeStatus(s *ServerStream, st *status.Status) error { if err != nil { return err } - t.closeStream(s, true, http2.ErrCodeInternal, false) + t.closeStream(s, true, http2.ErrCodeInternal, false, status.Convert(ErrHeaderListSizeLimitViolation)) return ErrHeaderListSizeLimitViolation } // Send a RST_STREAM after the trailers if the client has not already half-closed. @@ -1352,7 +1353,10 @@ func (t *http2Server) finishStream(s *ServerStream, rst bool, rstCode http2.ErrC } // closeStream clears the footprint of a stream when the stream is not needed any more. -func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCode, eosReceived bool) { +func (t *http2Server) closeStream(s *ServerStream, rst bool, rstCode http2.ErrCode, eosReceived bool, st *status.Status) { + // Record the status first so streamDone cannot be observed without its + // corresponding close status. + s.setCloseStatus(st) // In case stream sending and receiving are invoked in separate // goroutines (e.g., bi-directional streaming), cancel needs to be // called to interrupt the potential blocking on other goroutines. diff --git a/internal/transport/http_util.go b/internal/transport/http_util.go index faa42f8cad5c..8f2a1d349e79 100644 --- a/internal/transport/http_util.go +++ b/internal/transport/http_util.go @@ -40,6 +40,7 @@ import ( imem "google.golang.org/grpc/internal/mem" "google.golang.org/grpc/internal/transport/readyreader" "google.golang.org/grpc/mem" + "google.golang.org/grpc/status" ) const ( @@ -88,6 +89,14 @@ var ( } ) +func statusFromHTTP2Error(code http2.ErrCode, msg string) *status.Status { + statusCode, ok := http2ErrConvTab[code] + if !ok { + statusCode = codes.Unknown + } + return status.New(statusCode, msg) +} + var grpcStatusDetailsBinHeader = "grpc-status-details-bin" // isReservedHeader checks whether hdr belongs to HTTP2 headers diff --git a/internal/transport/server_stream.go b/internal/transport/server_stream.go index ed6a13b7501a..3ed98506ca97 100644 --- a/internal/transport/server_stream.go +++ b/internal/transport/server_stream.go @@ -30,6 +30,8 @@ import ( "google.golang.org/grpc/status" ) +const errSetSendCompressorTooLate = "transport: set send compressor called after headers sent or stream done" + // ServerStream implements streaming functionality for a gRPC server. type ServerStream struct { Stream // Embed for common stream functionality. @@ -48,6 +50,8 @@ type ServerStream struct { hdrMu sync.Mutex header metadata.MD // the outgoing header metadata. Updated by WriteHeader. headerSent atomic.Bool // atomically set when the headers are sent out. + // closeStatus is the first non-OK status recorded by closeStream. + closeStatus atomic.Pointer[status.Status] headerWireLength int } @@ -110,14 +114,25 @@ func (s *ServerStream) ContentSubtype() string { // SetSendCompress sets the compression algorithm to the stream. func (s *ServerStream) SetSendCompress(name string) error { - if s.isHeaderSent() || s.getState() == streamDone { - return errors.New("transport: set send compressor called after headers sent or stream done") + if s.isHeaderSent() { + return errors.New(errSetSendCompressorTooLate) + } + if st := s.closeStatus.Load(); st != nil { + return st.Err() + } + if s.getState() == streamDone { + return errors.New(errSetSendCompressorTooLate) } s.sendCompress = name return nil } +// setCloseStatus records the first status passed to closeStream. +func (s *ServerStream) setCloseStatus(st *status.Status) { + s.closeStatus.CompareAndSwap(nil, st) +} + // SetContext sets the context of the stream. This will be deleted once the // stats handler callouts all move to gRPC layer. func (s *ServerStream) SetContext(ctx context.Context) { diff --git a/internal/transport/server_stream_test.go b/internal/transport/server_stream_test.go new file mode 100644 index 000000000000..84fad547d66e --- /dev/null +++ b/internal/transport/server_stream_test.go @@ -0,0 +1,105 @@ +/* + * + * Copyright 2026 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package transport + +import ( + "context" + "testing" + + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +func (s) TestServerStreamSetSendCompressReturnsCloseStatus(t *testing.T) { + for _, test := range []struct { + name string + code codes.Code + }{ + {name: "canceled", code: codes.Canceled}, + {name: "deadline_exceeded", code: codes.DeadlineExceeded}, + {name: "internal", code: codes.Internal}, + {name: "resource_exhausted", code: codes.ResourceExhausted}, + } { + t.Run(test.name, func(t *testing.T) { + stream := &ServerStream{} + stream.setCloseStatus(status.New(test.code, "stream closed")) + + err := stream.SetSendCompress("gzip") + if got := status.Code(err); got != test.code { + t.Fatalf("SetSendCompress() returned code %v, want %v: %v", got, test.code, err) + } + }) + } +} + +func (s) TestServerStreamSetSendCompressAfterFinishedStream(t *testing.T) { + stream := &ServerStream{Stream: Stream{state: streamDone}} + + err := stream.SetSendCompress("gzip") + if got, want := err.Error(), errSetSendCompressorTooLate; got != want { + t.Fatalf("SetSendCompress() returned %q, want %q", got, want) + } + if _, ok := status.FromError(err); ok { + t.Fatalf("SetSendCompress() returned status error %v, want ordinary stream-done error", err) + } +} + +func (s) TestServerStreamSetSendCompressUsesFirstCloseStatus(t *testing.T) { + for _, test := range []struct { + name string + firstCode codes.Code + secondCode codes.Code + }{ + { + name: "internal_before_cancellation", + firstCode: codes.Internal, + secondCode: codes.Canceled, + }, + { + name: "cancellation_before_internal", + firstCode: codes.Canceled, + secondCode: codes.Internal, + }, + } { + t.Run(test.name, func(t *testing.T) { + stream := &ServerStream{} + stream.setCloseStatus(status.New(test.firstCode, "first close")) + stream.setCloseStatus(status.New(test.secondCode, "second close")) + + err := stream.SetSendCompress("gzip") + if got := status.Code(err); got != test.firstCode { + t.Fatalf("SetSendCompress() returned code %v, want first close code %v: %v", got, test.firstCode, err) + } + }) + } +} + +func (s) TestServerStreamSetSendCompressPrefersHeadersSent(t *testing.T) { + stream := &ServerStream{Stream: Stream{state: streamDone}} + stream.headerSent.Store(true) + stream.setCloseStatus(status.New(codes.Canceled, context.Canceled.Error())) + + err := stream.SetSendCompress("gzip") + if got, want := err.Error(), errSetSendCompressorTooLate; got != want { + t.Fatalf("SetSendCompress() returned %q, want %q", got, want) + } + if _, ok := status.FromError(err); ok { + t.Fatalf("SetSendCompress() returned status error %v, want ordinary headers-sent error", err) + } +} diff --git a/internal/transport/transport_test.go b/internal/transport/transport_test.go index 67a812e52fb1..d77a0fb7699c 100644 --- a/internal/transport/transport_test.go +++ b/internal/transport/transport_test.go @@ -101,6 +101,7 @@ type testStreamHandler struct { t *http2Server notify chan struct{} getNotified chan struct{} + streamCh chan *ServerStream } type hType int @@ -118,7 +119,10 @@ const ( pingpong ) -func (h *testStreamHandler) handleStreamAndNotify(*ServerStream) { +func (h *testStreamHandler) handleStreamAndNotify(s *ServerStream) { + if h.streamCh != nil { + h.streamCh <- s + } if h.notify == nil { return } @@ -372,6 +376,7 @@ type server struct { channelz *channelz.Server servingTasksDone chan struct{} timeout time.Duration + streamCh chan *ServerStream } func newTestServer() *server { @@ -429,7 +434,7 @@ func (s *server) start(t *testing.T, port int, serverConfig *ServerConfig, ht hT return } s.conns[transport] = rawConn - h := &testStreamHandler{t: transport.(*http2Server)} + h := &testStreamHandler{t: transport.(*http2Server), streamCh: s.streamCh} s.h = h s.mu.Unlock() timeout := s.timeout @@ -3448,8 +3453,12 @@ func (s) TestReadMessageHeaderMultipleBuffers(t *testing.T) { // configured deadline is reached. The test verifies that the server sends an // RST stream only after the deadline is reached. func (s) TestServerSendsRSTAfterDeadlineToMisbehavedClient(t *testing.T) { - server := setUpServerOnly(t, 0, &ServerConfig{BufferPool: mem.DefaultBufferPool()}, suspended) + server := setUpServerOnly(t, 0, &ServerConfig{BufferPool: mem.DefaultBufferPool()}, notifyCall) defer server.stop() + streamCh := make(chan *ServerStream, 1) + server.mu.Lock() + server.streamCh = streamCh + server.mu.Unlock() // Create a client that can override server stream quota. mconn, err := net.Dial("tcp", server.lis.Addr().String()) if err != nil { @@ -3518,6 +3527,13 @@ func (s) TestServerSendsRSTAfterDeadlineToMisbehavedClient(t *testing.T) { } mu.Unlock() + var serverStream *ServerStream + select { + case serverStream = <-streamCh: + case <-time.After(5 * time.Second): + t.Fatal("Timed out waiting for the server stream") + } + // Test server behavior for deadline expiration. var rstTime time.Time select { @@ -3529,6 +3545,13 @@ func (s) TestServerSendsRSTAfterDeadlineToMisbehavedClient(t *testing.T) { if got, want := rstTime.Sub(startTime), 10*time.Millisecond; got < want { t.Fatalf("RST frame received earlier than expected by duration: %v", want-got) } + st := serverStream.closeStatus.Load() + if st == nil { + t.Fatal("stream ended without a close status") + } + if got, want := st.Code(), codes.DeadlineExceeded; got != want { + t.Fatalf("stream ended with code %v, want %v", got, want) + } } // Tests the scenario where the client sends a DATA frame without END_STREAM @@ -4054,7 +4077,7 @@ func (s) TestDeleteStreamMetricsIncrementedOnlyOnce(t *testing.T) { // First call to closeStream should remove the stream from // the activeStreams and update metrics. closeStream will also // cancel the stream, stopping the deadline timer. - serverTransport.closeStream(serverStream, false, 0, test.eosReceived) + serverTransport.closeStream(serverStream, false, 0, test.eosReceived, status.New(codes.Canceled, "test closed stream")) // Check metrics after first deleteStream call streamsSucceeded := serverTransport.channelz.SocketMetrics.StreamsSucceeded.Load() diff --git a/server.go b/server.go index 1bc82131477b..a46d55598ccd 100644 --- a/server.go +++ b/server.go @@ -1412,6 +1412,9 @@ func (s *Server) processRPC(ctx context.Context, stream *transport.ServerStream, if ss.sendCompressorName != "" { if err := stream.SetSendCompress(ss.sendCompressorName); err != nil { + if st, ok := status.FromError(err); ok { + return st.Err() + } return status.Errorf(codes.Internal, "grpc: failed to set send compressor: %v", err) } } diff --git a/test/context_canceled_test.go b/test/context_canceled_test.go index 472f5b4cae3c..9341b505b09d 100644 --- a/test/context_canceled_test.go +++ b/test/context_canceled_test.go @@ -27,7 +27,9 @@ import ( "google.golang.org/grpc/codes" "google.golang.org/grpc/encoding/gzip" "google.golang.org/grpc/internal/stubserver" + "google.golang.org/grpc/internal/testutils" "google.golang.org/grpc/metadata" + "google.golang.org/grpc/stats" "google.golang.org/grpc/status" testgrpc "google.golang.org/grpc/interop/grpc_testing" @@ -162,3 +164,124 @@ func (s) TestCancelWhileRecvingWithCompression(t *testing.T) { t.Fatalf("Close failed with %v, want nil", err) } } + +// TestSendCompressorSetupDuringStreamClose verifies the status reported when +// response compressor setup races with the end of a compressed RPC. +func (s) TestSendCompressorSetupDuringStreamClose(t *testing.T) { + for _, test := range []struct { + name string + call func(context.Context, testgrpc.TestServiceClient) error + }{ + { + name: "unary", + call: func(ctx context.Context, client testgrpc.TestServiceClient) error { + _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.UseCompressor(gzip.Name)) + return err + }, + }, + { + name: "stream", + call: func(ctx context.Context, client testgrpc.TestServiceClient) error { + stream, err := client.FullDuplexCall(ctx, grpc.UseCompressor(gzip.Name)) + if err != nil { + return err + } + if err := stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil { + return err + } + _, err = stream.Recv() + return err + }, + }, + } { + for _, setupTest := range []struct { + name string + detachContext bool + }{ + { + name: "canceled_before_setup", + }, + { + name: "canceled_before_setup_with_detached_context", + detachContext: true, + }, + } { + t.Run(test.name+"/"+setupTest.name, func(t *testing.T) { + headerSeen := make(chan struct{}) + serverDone := make(chan error, 1) + handlerCalled := make(chan struct{}, 1) + var transportCtx context.Context + statsHandler := &testutils.StubStatsHandler{ + TagRPCF: func(ctx context.Context, _ *stats.RPCTagInfo) context.Context { + transportCtx = ctx + if setupTest.detachContext { + return context.WithoutCancel(ctx) + } + return ctx + }, + HandleRPCF: func(_ context.Context, stat stats.RPCStats) { + switch stat := stat.(type) { + case *stats.InHeader: + // Pause before send compressor setup until the stream is done. + close(headerSeen) + <-transportCtx.Done() + case *stats.End: + // Capture the server-side status observed by telemetry. + serverDone <- stat.Error + } + }, + } + + ss := &stubserver.StubServer{ + EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) { + handlerCalled <- struct{}{} + return &testpb.Empty{}, nil + }, + FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error { + handlerCalled <- struct{}{} + return nil + }, + } + if err := ss.Start([]grpc.ServerOption{grpc.StatsHandler(statsHandler)}); err != nil { + t.Fatalf("Error starting endpoint server: %v", err) + } + defer ss.Stop() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + callDone := make(chan error, 1) + go func() { + callDone <- test.call(ctx, ss.Client) + }() + + select { + case <-headerSeen: + case <-time.After(defaultTestTimeout): + t.Fatal("Timed out waiting for server to receive request headers") + } + cancel() + + select { + case <-callDone: + case <-time.After(defaultTestTimeout): + t.Fatal("Timed out waiting for RPC to finish") + } + + select { + case err := <-serverDone: + if got, want := status.Code(err), codes.Canceled; got != want { + t.Fatalf("Server stats ended with code %v, want %v: %v", got, want, err) + } + case <-time.After(defaultTestTimeout): + t.Fatal("Timed out waiting for server stats") + } + + select { + case <-handlerCalled: + t.Fatal("RPC handler was called after the request was canceled") + default: + } + }) + } + } +}