server: Preserve the RPC status when a compressed RPC ends before response compressor setup.

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.

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

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) {

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)
+	}
+}

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()

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)
 		}
 	}