grpc: place defaultStreamInterceptor after user interceptors (grpc#9404)

The `defaultStreamInterceptor` was introduced in grpc#9226.

This change places the `defaultStreamInterceptor` which performs certain
common logic for all non-unary RPCs to live after the user interceptors
and before the xDS filters. This ensures that the work of the
`defaultStreamInterceptor` stays invisible to the user interceptors.
Specifically, it calls `RecvMsg` twice for non server-streaming RPCs and
calls `CloseSend` after sending the only message for non
client-streaming RPCs.

RELEASE NOTES: none

---------

Co-authored-by: eshitachandwani <emchandwani@google.com>
(cherry picked from commit 2fbc883)

Author: 2 people

clientconn.go

@@ -553,7 +553,7 @@ func chainStreamClientInterceptors(cc *ClientConn) {
 	if cc.dopts.streamInt != nil {
 		interceptors = append([]StreamClientInterceptor{cc.dopts.streamInt}, interceptors...)
 	}
-	interceptors = append([]StreamClientInterceptor{defaultStreamInterceptor}, interceptors...)
+	interceptors = append(interceptors, defaultStreamInterceptor)
 	var chainedInt StreamClientInterceptor
 	if len(interceptors) == 0 {
 		chainedInt = nil

stream.go

@@ -153,7 +153,18 @@ type ClientStream interface {
 // RecvMsg parities based on the nature of stream.
 type clientStreamWrapper struct {
 	ClientStream
-	desc *StreamDesc
+	desc            *StreamDesc
+	closeSendCalled atomic.Bool
+}
+
+// CloseSend closes the send direction of the stream. The implementation ensures
+// that CloseSend is only called once on the underlying ClientStream, even if
+// CloseSend is called multiple times on the wrapper.
+func (w *clientStreamWrapper) CloseSend() error {
+	if w.closeSendCalled.Swap(true) {
+		return nil
+	}
+	return w.ClientStream.CloseSend()
 }
 
 // SendMsg sends message m across the stream. For RPCs where client can call
@@ -177,11 +188,13 @@ func (w *clientStreamWrapper) SendMsg(m any) error {
 	if err != nil {
 		return err
 	}
-	// CloseSend is needed because in some scenarios (e.g., xDS), the same
-	// interceptors are used to process both unary and streaming RPCs. Calling
-	// CloseSend signals to those interceptors that no more messages are on the
-	// way.
-	if err := w.ClientStream.CloseSend(); err != nil && err != io.EOF {
+	// In some scenarios (e.g., xDS), the same interceptors process both unary and
+	// streaming RPCs, relying on CloseSend to signal that no more messages are on
+	// the way. Although protobuf-generated stubs already invoke CloseSend for
+	// server-streaming RPCs, it is explicitly called here to ensure downstream
+	// interceptors are also notified when callers interact with the ClientStream
+	// API directly.
+	if err := w.CloseSend(); err != nil && err != io.EOF {
 		return err
 	}
 	return nil

stream_test.go

@@ -151,8 +151,8 @@ func (s) TestUnaryClient_ServerStreamingMismatch(t *testing.T) {
 // CloseSend hooks across downstream interceptors.
 type interceptorStream struct {
 	grpc.ClientStream
-	recvMsgCount int
-	closeSend    bool
+	recvMsgCount   int
+	closeSendCount int
 }
 
 func (s *interceptorStream) RecvMsg(m any) error {
@@ -161,14 +161,15 @@ func (s *interceptorStream) RecvMsg(m any) error {
 }
 
 func (s *interceptorStream) CloseSend() error {
-	s.closeSend = true
+	s.closeSendCount++
 	return s.ClientStream.CloseSend()
 }
 
-// TestDefaultStreamInterceptor verifies that defaultStreamInterceptor
-// automatically triggers CloseSend on non-client-streaming RPCs right after
-// SendMsg, and calls RecvMsg a second time on non-server-streaming RPCs to
-// consume trailers and io.EOF.
+// TestDefaultStreamInterceptor verifies that defaultStreamInterceptor's
+// behavior of automatically triggering CloseSend on non-client-streaming RPCs
+// right after SendMsg, and calling RecvMsg a second time on
+// non-server-streaming RPCs to consume trailers and io.EOF, are not visible to
+// user-defined interceptors.
 func (s) TestDefaultStreamInterceptor(t *testing.T) {
 	var iStream *interceptorStream
 	// Define a client-side stream interceptor that wraps the ClientStream to
@@ -207,10 +208,11 @@ func (s) TestDefaultStreamInterceptor(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
 
-	// Make a client-streaming RPC. When CloseAndRecv invokes RecvMsg once to get
-	// the single reply message on a non-server-streaming RPC,
+	// Make a client-streaming RPC. When CloseAndRecv invokes RecvMsg once to
+	// get the single reply message on a non-server-streaming RPC,
 	// defaultStreamInterceptor automatically calls a second RecvMsg on the
-	// underlying client stream to consume io.EOF and receive trailers.
+	// underlying client stream to consume io.EOF and receive trailers. But this
+	// should not be visible to user interceptors.
 	stream, err := ss.Client.StreamingInputCall(ctx)
 	if err != nil {
 		t.Fatal("Error calling StreamingInputCall:", err)
@@ -221,17 +223,21 @@ func (s) TestDefaultStreamInterceptor(t *testing.T) {
 	if _, err := stream.CloseAndRecv(); err != nil {
 		t.Fatal("Error running CloseAndRecv:", err)
 	}
-	if iStream.recvMsgCount != 2 {
-		t.Fatalf("StreamingInputCall RecvMsg was called %v times, want 2 times", iStream.recvMsgCount)
+	if iStream.recvMsgCount != 1 {
+		t.Fatalf("RecvMsg was called %v times on user interceptor stream, want 1 time", iStream.recvMsgCount)
 	}
 
 	// Make a server-streaming RPC. Since StreamingOutputCall is not
-	// client-streaming, defaultStreamInterceptor immediately invokes CloseSend
-	// right after sending the request message to signal downstream interceptors.
+	// client-streaming, the proto generated code invokes CloseSend after
+	// sending the single message. defaultStreamInterceptor also invokes
+	// CloseSend right after sending the request message (to handle cases where
+	// the user is using the ClientStream API instead of the proto generated
+	// code) to signal downstream interceptors, which in this case are xDS
+	// filters. The user interceptor should not see this CloseSend call.
 	if _, err := ss.Client.StreamingOutputCall(ctx, &testpb.StreamingOutputCallRequest{}); err != nil {
 		t.Fatal("Error calling StreamingOutputCall:", err)
 	}
-	if !iStream.closeSend {
-		t.Fatal("CloseSend not called after SendMsg on non-client-streaming RPC")
+	if iStream.closeSendCount != 1 {
+		t.Fatalf("CloseSend called %v times on user interceptor stream, want 1 times", iStream.closeSendCount)
 	}
 }

test/xds/xds_client_filter_e2e_test.go

@@ -0,0 +1,239 @@
+/*
+ *
+ * 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 xds_test
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"testing"
+
+	v3xdsxdstypepb "github.com/cncf/xds/go/xds/type/v3"
+	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
+	v3endpointpb "github.com/envoyproxy/go-control-plane/envoy/config/endpoint/v3"
+	v3listenerpb "github.com/envoyproxy/go-control-plane/envoy/config/listener/v3"
+	v3routepb "github.com/envoyproxy/go-control-plane/envoy/config/route/v3"
+	v3httppb "github.com/envoyproxy/go-control-plane/envoy/extensions/filters/network/http_connection_manager/v3"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/stubserver"
+	"google.golang.org/grpc/internal/testutils"
+	"google.golang.org/grpc/internal/testutils/xds/e2e"
+	"google.golang.org/grpc/internal/testutils/xds/e2e/setup"
+	"google.golang.org/grpc/internal/xds/httpfilter"
+
+	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
+)
+
+// Tests the interaction between the defaultStreamInterceptor and xDS filters.
+// It verifies that the SendMsg, RecvMsg, and CloseSend methods are called the
+// expected number of times for different RPC types (unary, client streaming,
+// server streaming, and bidi streaming) when an xDS filter is registered and
+// used in the call chain.
+func (s) TestDefaultStreamInterceptor_InteractionWithXDSFilters(t *testing.T) {
+	// Register a custom xDS filter builder for the test.
+	testFilterTypeURL := t.Name()
+	filterBuilder := newTrackingHTTPFilterBuilder(testFilterTypeURL)
+	httpfilter.Register(filterBuilder)
+	defer httpfilter.UnregisterForTesting(testFilterTypeURL)
+
+	// Setup a test backend implementing both all four RPC types.
+	testServer := &stubserver.StubServer{
+		EmptyCallF: func(context.Context, *testpb.Empty) (*testpb.Empty, error) {
+			return &testpb.Empty{}, nil
+		},
+		StreamingOutputCallF: func(_ *testpb.StreamingOutputCallRequest, stream testgrpc.TestService_StreamingOutputCallServer) error {
+			return stream.Send(&testpb.StreamingOutputCallResponse{})
+		},
+		StreamingInputCallF: func(stream testgrpc.TestService_StreamingInputCallServer) error {
+			for {
+				_, err := stream.Recv()
+				if err == io.EOF {
+					break
+				}
+				if err != nil {
+					return err
+				}
+			}
+			return stream.SendAndClose(&testpb.StreamingInputCallResponse{})
+		},
+		FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+			for {
+				_, err := stream.Recv()
+				if err == io.EOF {
+					return nil
+				}
+				if err != nil {
+					return err
+				}
+				if err := stream.Send(&testpb.StreamingOutputCallResponse{}); err != nil {
+					return err
+				}
+			}
+		},
+	}
+	if err := testServer.Start(nil); err != nil {
+		t.Fatal("Error starting server:", err)
+	}
+	defer testServer.Stop()
+
+	// Start an xDS management server.
+	mgmtServer, nodeID, _, xdsResolver := setup.ManagementServerAndResolver(t)
+
+	const serviceName = "my-service-xds"
+	clusterSpec := &v3routepb.RouteAction_Cluster{Cluster: "cluster-A"}
+	hcm := &v3httppb.HttpConnectionManager{
+		RouteSpecifier: &v3httppb.HttpConnectionManager_RouteConfig{
+			RouteConfig: &v3routepb.RouteConfiguration{
+				Name: "route-" + serviceName,
+				VirtualHosts: []*v3routepb.VirtualHost{{
+					Domains: []string{serviceName},
+					Routes: []*v3routepb.Route{{
+						Match: &v3routepb.RouteMatch{PathSpecifier: &v3routepb.RouteMatch_Prefix{Prefix: ""}},
+						Action: &v3routepb.Route_Route{Route: &v3routepb.RouteAction{
+							ClusterSpecifier: clusterSpec,
+						}},
+					}},
+				}},
+			},
+		},
+		HttpFilters: []*v3httppb.HttpFilter{
+			{
+				Name: "tracking-filter",
+				ConfigType: &v3httppb.HttpFilter_TypedConfig{
+					TypedConfig: testutils.MarshalAny(t, &v3xdsxdstypepb.TypedStruct{
+						TypeUrl: testFilterTypeURL,
+					}),
+				},
+			},
+			e2e.RouterHTTPFilter,
+		},
+	}
+	resources := e2e.UpdateOptions{
+		NodeID:    nodeID,
+		Listeners: []*v3listenerpb.Listener{{Name: serviceName, ApiListener: &v3listenerpb.ApiListener{ApiListener: testutils.MarshalAny(t, hcm)}}},
+		Clusters:  []*v3clusterpb.Cluster{e2e.DefaultCluster("cluster-A", "cluster-A", e2e.SecurityLevelNone)},
+		Endpoints: []*v3endpointpb.ClusterLoadAssignment{e2e.DefaultEndpoint("cluster-A", "localhost", []uint32{testutils.ParsePort(t, testServer.Address)})},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	if err := mgmtServer.Update(ctx, resources); err != nil {
+		t.Fatal(err)
+	}
+
+	cc, err := grpc.NewClient(fmt.Sprintf("xds:///%s", serviceName), grpc.WithTransportCredentials(insecure.NewCredentials()), grpc.WithResolvers(xdsResolver))
+	if err != nil {
+		t.Fatalf("Failed to create a gRPC client: %v", err)
+	}
+	defer cc.Close()
+	client := testgrpc.NewTestServiceClient(cc)
+
+	tests := []struct {
+		name        string
+		run         func(ctx context.Context, client testgrpc.TestServiceClient)
+		wantSendMsg int32
+		wantRecvMsg int32
+	}{
+		{
+			name: "unary",
+			run: func(ctx context.Context, client testgrpc.TestServiceClient) {
+				if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+					t.Fatalf("EmptyCall() failed: %v", err)
+				}
+			},
+			wantSendMsg: 1, // 1 request
+			wantRecvMsg: 2, // 1 response + 1 trailers
+		},
+		{
+			name: "client_streaming",
+			run: func(ctx context.Context, client testgrpc.TestServiceClient) {
+				stream, err := client.StreamingInputCall(ctx)
+				if err != nil {
+					t.Fatalf("StreamingInputCall() failed: %v", err)
+				}
+				for i := 0; i < 2; i++ {
+					if err := stream.Send(&testpb.StreamingInputCallRequest{}); err != nil {
+						t.Fatalf("Send() failed: %v", err)
+					}
+				}
+				if _, err := stream.CloseAndRecv(); err != nil {
+					t.Fatalf("CloseAndRecv() failed: %v", err)
+				}
+			},
+			wantSendMsg: 2, // 2 requests
+			wantRecvMsg: 2, // 1 response + 1 trailers
+		},
+		{
+			name: "server_streaming",
+			run: func(ctx context.Context, client testgrpc.TestServiceClient) {
+				stream, err := client.StreamingOutputCall(ctx, &testpb.StreamingOutputCallRequest{})
+				if err != nil {
+					t.Fatalf("StreamingOutputCall() failed: %v", err)
+				}
+				if _, err := stream.Recv(); err != nil {
+					t.Fatalf("Recv() failed: %v", err)
+				}
+			},
+			wantSendMsg: 1, // 1 request
+			wantRecvMsg: 1, // 1 response (trailers not yet consumed since stream not read to EOF)
+		},
+		{
+			name: "bidi_streaming",
+			run: func(ctx context.Context, client testgrpc.TestServiceClient) {
+				stream, err := client.FullDuplexCall(ctx)
+				if err != nil {
+					t.Fatalf("FullDuplexCall() failed: %v", err)
+				}
+				if err := stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
+					t.Fatalf("Send() failed: %v", err)
+				}
+				if _, err := stream.Recv(); err != nil {
+					t.Fatalf("Recv() failed: %v", err)
+				}
+				if err := stream.CloseSend(); err != nil {
+					t.Fatalf("CloseSend() failed: %v", err)
+				}
+			},
+			wantSendMsg: 1, // 1 request
+			wantRecvMsg: 1, // 1 response
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			filterBuilder.sendMsgCount.Store(0)
+			filterBuilder.recvMsgCount.Store(0)
+			filterBuilder.closeSendCount.Store(0)
+
+			tc.run(ctx, client)
+
+			if got := filterBuilder.sendMsgCount.Load(); got != tc.wantSendMsg {
+				t.Fatalf("SendMsg() count = %d, want %d", got, tc.wantSendMsg)
+			}
+			if got := filterBuilder.recvMsgCount.Load(); got != tc.wantRecvMsg {
+				t.Fatalf("RecvMsg() count = %d, want %d", got, tc.wantRecvMsg)
+			}
+			if got := filterBuilder.closeSendCount.Load(); got != 1 {
+				t.Fatalf("CloseSend() count = %d, want 1", got)
+			}
+		})
+	}
+}